Java Spark 无法处理整个文本文件

Question

我是 Spark API 的新手并一路学习。

我在hadoop目录中有多个文件，我正在使用wholeTextFiles读取这些文件来创建JavaPairRDD<String, String> 。程序是用 Java 编写的。

我的要求是处理目录中的文件列表并实现以下输出:

file-path, word

file-path, word

file-path, word

...

这基本上是文件的文字内容，相应的文件名(或路径)配对为 <String, String> .

我尝试了以下操作，但是不允许从 tuple2 到 Iterable 的转换(在运行时失败):

JavaSparkContext sc = new JavaSparkContext(new SparkConf()) ;

JavaPairRDD<String, String> files = sc.wholeTextFiles(args[0]);

JavaRDD<Tuple2<String, String>> file_word = files
.flatMap(new FlatMapFunction<Tuple2<String,String>, Tuple2<String,String>>()
{
public Iterable<Tuple2<String, String>> call(Tuple2<String, String> tuple) 
{
 return (Iterable<Tuple2<String, String>>) new Tuple2<String, Iterable<String>>(tuple._1(), Arrays.asList(((String) tuple._2()).toLowerCase ().split("\\W+")));
}
});

我正在使用Java 8 , Hadoop2与 Spark 2.2.0 .

(通过查看这里的其他问题，我可以理解用 scala 编写这个更容易，但是我没有找到 Java 的相关答案)

正在寻找解决方案。谢谢。

Answer 1

据我所知，您试图将 Tuple2 转换为 Iterable，但这是行不通的。

由于您使用的是 java8，因此您可以使用 lambda 表达式来编写此表达式，这将使事情变得更加紧凑:

JavaPairRDD<String, String> rdd = sc
            .wholeTextFiles("path_to_data/*")
            .flatMapValues(x -> Arrays.asList(x.split("\\W+")));

请注意，我使用 flatMapValues 而不是 flatMap，因为您只需要处理元组的第二个值。

如果您好奇，使用 flatmap 您可以通过将文件的每个单词映射到一个元组(文件名，单词)来完成此操作:

JavaRDD<Tuple2<String, String>> rdd2 = sc
            .wholeTextFiles("path_to_data/*")
            .flatMap(x -> Arrays.asList(x._2.split("\\n"))
                    .stream()
                    .map(w -> new Tuple2<>(x._1, w))
                    .iterator());

flatMapValues 只需让您用更少的代码即可做到这一点;-)