map，filter，flatMap算子
视频教程：
1、优酷
2、YouTube
1、map
map是将源JavaRDD的一个一个元素的传入call方法，并经过算法后一个一个的返回从而生成一个新的JavaRDD。
java：
package com.bean.spark.trans; import java.util.Arrays;
import java.util.List;
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;
/**
*
* @author RedBean
*map
*/
public class TraMap {
public static void main(String[] args) {
SparkConf conf = new SparkConf();
conf.setMaster("local");
conf.setAppName("map");
System.setProperty("hadoop.home.dir", "D:/tools/spark-2.0.0-bin-hadoop2.6");
JavaSparkContext sc = new JavaSparkContext(conf);
List<Integer> number = Arrays.asList(0,1,2,3,4,5,6,7,8,9);
JavaRDD<Integer> numberRDD = sc.parallelize(number);
JavaRDD<Integer> results = numberRDD.map(new Function<Integer, Integer>() {
@Override
public Integer call(Integer s) throws Exception {
// TODO Auto-generated method stub
return s * 5;
}
});
System.out.println(results.collect());
}
}
python：
# -*- coding:utf-8 -*- from __future__ import print_function
from pyspark import SparkConf
from pyspark import SparkContext
import os if __name__ == '__main__':
os.environ['SPARK_HOME'] = 'D:/tools/spark-2.0.0-bin-hadoop2.6'
conf = SparkConf().setAppName('mapTest').setMaster('local')
sc = SparkContext(conf=conf)
data = sc.parallelize([1,2,3,4,5,6])
def myMap(l):
return l * 5
print(data.map(myMap).collect())
2、filter
返回一个新的数据集，由经过func函数后返回值为true的原元素组成
java：
package com.bean.spark.trans; import java.util.Arrays;
import java.util.List; 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 TraFilter {
public static void main(String[] args) {
SparkConf conf = new SparkConf();
conf.setMaster("local");
conf.setAppName("filter");
System.setProperty("hadoop.home.dir", "D:/tools/spark-2.0.0-bin-hadoop2.6");
JavaSparkContext sc = new JavaSparkContext(conf);
List<Integer> number = Arrays.asList(0,1,2,3,4,5,6,7,8,9);
JavaRDD<Integer> numberRDD = sc.parallelize(number);
JavaRDD<Integer> results = numberRDD.filter(new Function<Integer, Boolean>() { @Override
public Boolean call(Integer s) throws Exception {
// TODO Auto-generated method stub
return s % 2 == 0;
}
});
System.out.println(results.collect());
}
}
python：
# -*- coding:utf-8 -*- from __future__ import print_function
from pyspark import SparkConf
from pyspark import SparkContext
import os if __name__ == '__main__':
os.environ['SPARK_HOME'] = 'D:/tools/spark-2.0.0-bin-hadoop2.6'
conf = SparkConf().setAppName('filterTest').setMaster('local')
sc = SparkContext(conf=conf)
data = sc.parallelize([1,2,3,4,5,6])
def filterFun(l):
return l > 2
print(data.filter(filterFun).collect())
3、flatMap
将一条 rdd数据使用你定义的函数给分解成多条 rdd数据。
java：
package com.bean.spark.trans; import java.util.Arrays;
import java.util.Iterator; 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.FlatMapFunction; public class TraFlatMap {
public static void main(String[] args) {
SparkConf conf = new SparkConf();
conf.setMaster("local");
conf.setAppName("FlatMap");
System.setProperty("hadoop.home.dir", "D:/tools/spark-2.0.0-bin-hadoop2.6");
JavaSparkContext sc = new JavaSparkContext(conf);
JavaRDD<String> context = sc.textFile("D:/tools/data/flatMap/flatMap.txt");
JavaRDD<String> results = context.flatMap(new FlatMapFunction<String, String>() {
@Override
public Iterator<String> call(String s) throws Exception {
// TODO Auto-generated method stub
return Arrays.asList(s).iterator();
}
});
System.out.println(results.collect()); }
}
python：
# -*- coding:utf-8 -*- from __future__ import print_function
from pyspark import SparkConf
from pyspark import SparkContext
import os if __name__ == '__main__':
os.environ['SPARK_HOME'] = 'D:/tools/spark-2.0.0-bin-hadoop2.6'
conf = SparkConf().setAppName('filterTest').setMaster('local')
sc = SparkContext(conf=conf)
data = sc.parallelize(["Hello World","Spark Hadoop Storm","java python c"])
def flatFun(l):
return l.split(" ")
print(data.flatMap(flatFun).collect())