使用Java8 Lambda表达式进行Spark编程

写两个例子,用于说明Java 8如何使代码更简洁。第一个例子是使用Spark的filter和count算子在一个日志文件中查找包含“error”的行。这很容易实现,但在Java 7中需要向filter传递一个Function对象,这有些笨拙:

JavaRDD lines = sc.textFile("hdfs://log.txt").filter( new Function() { public Boolean call(String s) { return s.contains("error"); } }); long numErrors = lines.count();

在Java 8中,代码更为简洁:
JavaRDD lines = sc.textFile("hdfs://log.txt") .filter(s -> s.contains("error")); long numErrors = lines.count();

当代码更长时,对比更明显。文中给出了第二个例子,读取一个文件,得出其中的单词数。在Java 7中,实现代码如下:
JavaRDD lines = sc.textFile("hdfs://log.txt"); //将每一行映射成多个单词 JavaRDD words = lines.flatMap( new FlatMapFunction() { public Iterable call(String line) { return Arrays.asList(line.split(" ")); } }); // 将单词转换成(word, 1)对 JavaPairRDD ones = words.mapToPair( new PairFunction() { public Tuple2 call(String w) { return new Tuple2(w, 1); } }); // 分组并按键值添加对以产生计数 JavaPairRDD counts = ones.reduceByKey( new Function2() { public Integer call(Integer i1, Integer i2) { return i1 + i2; } }); counts.saveAsTextFile("hdfs://counts.txt");

而在Java 8中,该程序只需要几行代码:
JavaRDD lines = sc.textFile("hdfs://log.txt"); JavaRDD words = lines.flatMap(line -> Arrays.asList(line.split(" "))); JavaPairRDD counts = words.mapToPair(w -> new Tuple2(w, 1)) .reduceByKey((x, y) -> x + y); counts.saveAsTextFile("hdfs://counts.txt");

要了解更多关于Spark的信息,可以查看官方文档。Spark只需下载解压即可运行,而无须安装。
注:
最后这个地方,先生成lines,再生成words,代码正常运行,之前从网上其他地方粘贴过来的代码都是下面这样:
【使用Java8 Lambda表达式进行Spark编程】lines.flatMap(line -> Arrays.asList(line.split("\\001")).iterator()).foreach(word -> System.out.println(word));

结果eclipse总是报错,报错意思是iterator是String类型的,前面的不是,改了好几个小时,也没找到原因,最后这篇文章帮我解决了问题。
使用Java8 Lambda表达式进行Spark编程
文章图片


    推荐阅读