Отображение из String в Tuple2в Spark + Java

Question

Я пытаюсь узнать, как использовать Spark, кодирование на Java (пожалуйста, не используйте код Scala).Я пытаюсь реализовать очень простой привет мир пример Spark, подсчет слов.

Я позаимствовал код из документации Spark быстрый старт :

/* SimpleApp.java */
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.Dataset;

public class SimpleApp {
  public static void main(String[] args) {
    String logFile = "YOUR_SPARK_HOME/README.md"; // Should be some file on your system
    SparkSession spark = SparkSession.builder().appName("Simple Application").getOrCreate();
    Dataset<String> logData = spark.read().textFile(logFile).cache();

    long numAs = logData.filter(s -> s.contains("a")).count();
    long numBs = logData.filter(s -> s.contains("b")).count();

    System.out.println("Lines with a: " + numAs + ", lines with b: " + numBs);

    spark.stop();
  }
}

Все хорошо, теперь я хочу заменить filter на flatMap, а затем map.Пока у меня есть flatMap:

    logData.flatMap((FlatMapFunction<String, String>) l -> {
                          return Arrays.asList(l.split(" ")).iterator();
                    }, Encoders.STRING());

Теперь я хочу сопоставить каждое слово с Tuple2 (word, 1), а затем сгруппировать их по ключу.Но проблема в том, что я не могу найти, как добраться от String до (String, Long).В большинстве документов говорится о mapToPair, но Dataset не имеет такого метода!

Может ли кто-нибудь помочь мне сопоставить String с Tuple2<String, Long>?Кстати, я даже не уверен, ищу ли я Tuple2 или какой-то другой класс.

[ОБНОВЛЕНИЕ]

Основано на предложении, предоставленном @mangusta, я попробовал это:

    logData.flatMap((FlatMapFunction<String, String>) l -> {
        return Arrays.asList(l.split(" ")).iterator();
    }, Encoders.STRING())
    .map(new Function<String, Tuple2<String, Long>>() {
        public Tuple2<String, Long> call(String str) {
            return new Tuple2<String, Long>(str, 1L);
        }
    })
    .count()

И столкнулся с этой ошибкой компиляции:

Error:(108, 17) java: no suitable method found for map(<anonymous org.apache.spark.api.java.function.Function<java.lang.String,scala.Tuple2<java.lang.String,java.lang.Long>>>)
    method org.apache.spark.sql.Dataset.<U>map(scala.Function1<java.lang.String,U>,org.apache.spark.sql.Encoder<U>) is not applicable
      (cannot infer type-variable(s) U
        (actual and formal argument lists differ in length))
    method org.apache.spark.sql.Dataset.<U>map(org.apache.spark.api.java.function.MapFunction<java.lang.String,U>,org.apache.spark.sql.Encoder<U>) is not applicable
      (cannot infer type-variable(s) U
        (actual and formal argument lists differ in length))

Похоже, функция map принимает два параметра.Я не уверен, что я должен передать в качестве второго параметра.

Answer 1

Я не уверен, что причина ошибки, но вы можете попробовать этот код

final String sparkHome = "/usr/local/Cellar/apache-spark/2.3.2";
SparkConf conf = new SparkConf()
        .setMaster("local[*]")
        .setAppName("spark-example")
        .setSparkHome(sparkHome + "/libexec");

SparkSession spark = SparkSession.builder().config(conf).getOrCreate();
Dataset<Row> df = spark.read().textFile(sparkHome + "/README.md")
        .flatMap(line -> Arrays.asList(line.split(" ")).iterator(), Encoders.STRING())
        .filter(s -> !s.isEmpty())
        .map(word -> new Tuple2<>(word.toLowerCase(), 1L), Encoders.tuple(Encoders.STRING(), Encoders.LONG()))
        .toDF("word", "count")
        .groupBy("word")
        .sum("count").orderBy(new Column("sum(count)").desc()).withColumnRenamed("sum(count)", "_cnt");

df.show(false);

И вы должны ожидать этот вывод

+-------------+----+
|word         |_cnt|
+-------------+----+
|the          |25  |
|to           |19  |
|spark        |16  |
|for          |15  |
|and          |10  |
|a            |9   |
|##           |9   |
|you          |8   |
|run          |7   |
|on           |7   |
|can          |7   |
|is           |6   |
|in           |6   |
|of           |5   |
|using        |5   |
|including    |4   |
|if           |4   |
|with         |4   |
|documentation|4   |
|an           |4   |
+-------------+----+
only showing top 20 rows

Answer 2

Если вам нужно использовать Tuple2, вам следует использовать библиотеку Scala для Java, т.е. scala-library.jar

Чтобы подготовить кортежи из некоторых JavaRDD<String> data, вы можете применить к этому СДР следующую функцию:

JavaRDD<Tuple2<String,Long>> tupleRDD  =  data.map(

new Function<String, Tuple2<String, Long>>() {

            public Tuple2<String, Long> call(String str) {

              return new Tuple2<String, Long>(str, 1L);

            }//end call

          }//end function

        );//end map