spark map flatMap flatMapToPair mapPartitions 的区别和用途

import akka.japi.Function2;
import org.apache.spark.HashPartitioner;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.storage.StorageLevel;
import scala.Tuple2;

import java.io.File;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

/**
* map flatMap flatMapToPair mapPartitions 的区别和用途
*
* 例如数据是：name：gaoyue age：28
*
* 方法一：map,我们可以看到数据的每一行在map之后产生了一个数组，那么rdd存储的是一个数组的集合
* rdd存储的状态是Array[Array[String]] = Array(Array(name, gaoyue), Array(age, 28))
*Array[String] = Array(name, gaoyue, age, 28)
*/

JavaRDD<String[]> mapresult=lines.map(new Function<String, String[]>() {

@Override
public String[] call(String s) throws Exception {
return s.split(":");
}
});

/**
* 方法二：flatMap
* 操作1：同map函数一样：对每一条输入进行指定的操作，然后为每一条输入返回一个对象
* 操作2：最后将所有对象合并为一个对象
*/
JavaRDD<String> objectJavaRDD = lines.flatMap(new FlatMapFunction<String, String>() {

@Override
public Iterable<String> call(String s) throws Exception {
return Arrays.asList(s.split(" "));
}
});

/**
* 方法三：
* mappartition
*rdd的mapPartitions是map的一个变种，它们都可进行分区的并行处理。两者的主要区别是调用的粒度不一样：
* map的输入变换函数是应用于RDD中每个元素，而mapPartitions的输入函数是应用于每个分区。也就是把每个分区中的内容作为整体来处理的。
*
*/
lines2.mapPartitions(new FlatMapFunction<Iterator<String>, String>() {
ArrayList<String> results = new ArrayList<String>();

@Override
public Iterable<String> call(Iterator<String> s) throws Exception {
while (s.hasNext()) {
results.addAll(Arrays.asList(s.next().split(":")));
}
return results;
}
}).saveAsTextFile("/Users/luoluowushengmimi/Documents/result");

/**
* flatMapToPair
* 操作1：同map函数一样：对每一条输入进行指定的操作，然后为每一条输入返回一个key－value对象
* 操作2：最后将所有key－value对象合并为一个对象 Iterable<Tuple2<String, String>>
*
*/

JavaPairRDD<String,String> pair=lines.flatMapToPair(new PairFlatMapFunction<String, String, String>() {

@Override
public Iterable<Tuple2<String, String>> call(String s) throws Exception {
String[] temp=s.split(":");
ArrayList<Tuple2<String,String>> list=new ArrayList<Tuple2<String,String>>();
list.add(new Tuple2<String,String>(temp[0],temp[1]));
return list;
}
});