本篇分析一个spark例子程序。

程序实现的功能是：分别统计包含字符a、b的行数。

java源码如下：

package sparkTest;

import org.apache.spark.SparkConf;

import org.apache.spark.api.java.JavaRDD;

import org.apache.spark.api.java.JavaSparkContext;

import org.apache.spark.api.java.function.Function;

public class SimpleApp {

public static void main(String[] args) {

String logFile = "file:///usr/local/spark/README.md"; // Should be some file on your system

SparkConf conf = new SparkConf().setAppName("Simple Application").setMaster("local");

JavaSparkContext sc = new JavaSparkContext(conf);

JavaRDD logData = sc.textFile(logFile).cache();//将文件cache在内存中

long numAs = logData.filter(new Function() {//保留包含a的元素

public Boolean call(String s) {

return s.contains("a");//

}

}).count();//RDD元素总数

long numBs = logData.filter(new Function() {//String为输入类型，保留Boolean为true元素

public Boolean call(String s) {

return s.contains("b");

}

}).count();

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

}

}

基本步骤解析如下：

1、设置sparkContext，包括sparkConf，读取文件的地址和协议。

2、将文件内容cache()在内存中。经过实验，不cache()也能执行显示正确结果。

3、JavaRDD.filter()。对元素进行过滤，过滤的方法是函数Function(String, Boolean)。

具体定义在call(String s)中。

其中Function的参数String与call的参数String对应。Function的参数Boolean与call函数返回值Boolean类型对应。

返回Boolean为true的元素通过filter，保留在新生成的ＲＤＤ中。

4、JavaRDD.count()对RDD元素个数进行统计。

5、最后，打印结果。

注意：这里的filter()、count()对应的就是transformation、action。