向Spark传递函数

Spark中的大部分转化操作和一部分行动操作，都需要依赖用户传递的函数来进行计算。Java中，函数需要是实现了function包中的任意函数接口的对象，根据不同的返回类型，定义了一些不同的接口：

Function    R  call(T)   接收一个输入值，并返回一个输出值，用于类似map()和filter()方法  

Function2   R  call(T1, T2)  接收两个输入值并返回一个输出值，用于类似于aggregate()和fold()等操作  

FlatMapFunction  Iterable  call(T)  接收一个输入值并返回任意个输出，用于类似flatMap()  

可以通过匿名内部类，也可以使用具名类的方式

不过，顶级具名类在组织大型程序时显得比较清晰，另一个好处就是，可以给构造函数添加参数

// 匿名类  

JavaRDD input_filter = input.filter(new Function(){  

@Override  

public Boolean call(String arg0) throws Exception {  

return arg0.contains("\"4\"");  

    }  

});  

System.out.println(input_filter.count());//237 

//具名类  

class Contains implements Function{  

    String filterStr;  

public Contains(String str){  

        filterStr = str;  

    }  

@Override  

public Boolean call(String arg0) throws Exception {  

return arg0.contains(filterStr);  

    }  

} 

JavaRDD filters = input.filter(new Contains("\"9\""));  

System.out.println(filters.count());//637  

在Java 8 中，也可以使用lambda表达式进行函数传递

RDD str = lines.filter(s -> s.contains("error"));