Spark---转换算子、行动算子、持久化算子

一、转换算子和行动算子

1、Transformations转换算子

1）、概念

Transformations类算子是一类算子（函数）叫做转换算子，如map、flatMap、reduceByKey等。Transformations算子是延迟执行，也叫懒加载执行。

2)、Transformation类算子

filter :过滤符合条件的记录数，true保留，false过滤掉

map：将一个RDD中的每个数据项，通过map中的函数映射变为一个新的元素。特点：输入一条，输出一条数据。

flatMap：先map后flat。与map类似，每个输入项可以映射为0到多个输出项。

sample：随机抽样算子，根据传进去的小数按比例进行又放回或者无放回的抽样。

reduceByKey：将相同的Key根据相应的逻辑进行处理。

sortByKey/sortBy:作用在K，V格式的RDD上，对Key进行升序或者降序排序。

2、Action行动算子

1）、概念：

Action类算子也是一类算子（函数）叫做行动算子，如foreach,collect,count等。Transformations类算子是延迟执行，Action类算子是触发执行。一个application应用程序中有几个Action类算子执行，就有几个job运行。

2）、Action类算子

count：返回数据集中的元素数。会在结果计算完成后回收到Driver端。

take(n):返回一个包含数据集前n个元素的集合。

first：first=take(1),返回数据集中的第一个元素。

foreach：循环遍历数据集中的每个元素，运行相应的逻辑。

collect：将计算结果回收到Driver端。

3）、demo：动态统计出现次数最多的单词个数，过滤掉。

一千万条数据量的文件，过滤掉出现次数多的记录，并且其余记录按照出现次数降序排序。

假设有一个records.txt文件

hello Spark
hello HDFS
hello hadoop
hello linux
hello Spark
hello Spark
hello Spark1
hello Spark
hello Spark
hello Spark2
hello Spark
hello Spark
hello Spark
hello Spark3
hello Spark
hello HDFS
hello hadoop
hello linux
hello Spark
hello Spark
hello Spark4
hello Spark
hello Spark
hello Spark5
hello Spark
hello Spark

代码处理：

package com.bjsxt.demo;

import java.util.Arrays;
import java.util.List;

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.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;

import scala.Tuple2;
/**
 * 动态统计出现次数最多的单词个数，过滤掉。
 * @author root
 *
 */
public class Demo1 {
	public static void main(String[] args) {
		SparkConf conf = new SparkConf();
		conf.setMaster("local").setAppName("demo1");
		JavaSparkContext jsc = new JavaSparkContext(conf);
		JavaRDD<String> lines = jsc.textFile("./records.txt");
		JavaRDD<String> flatMap = lines.flatMap(new FlatMapFunction<String, String>() {

			/**
			 * 
			 */
			private static final long serialVersionUID = 1L;

			@Override
			public Iterable<String> call(String t) throws Exception {
				return Arrays.asList(t.split(" "));
			}
		});
		JavaPairRDD<String, Integer> mapToPair = flatMap.mapToPair(new PairFunction<String,String, Integer>() {

			/**
			 * 
			 */
			private static final long serialVersionUID = 1L;

			@Override
			public Tuple2<String, Integer> call(String t) throws Exception {
				return new Tuple2<String, Integer>(t, 1);
			}
		});
		
		JavaPairRDD<String, Integer> sample = mapToPair.sample(true, 0.5);
		
		final List<Tuple2<String, Integer>> take = sample.reduceByKey(new Function2<Integer,Integer,Integer>(){

			/**
			 * 
			 */
			private static final long serialVersionUID = 1L;

			@Override
			public Integer call(Integer v1, Integer v2) throws Exception {
				return v1+v2;
			}
			
		}).mapToPair(new PairFunction<Tuple2<String,Integer>, Integer, String>() {

			/**
			 * 
			 */
			private static final long serialVersionUID = 1L;

			@Override
			public Tuple2<Integer, String> call(Tuple2<String, Integer> t)
					throws Exception {
				return new Tuple2<Integer, String>(t._2, t._1);
			}
		}).sortByKey(false).mapToPair(new PairFunction<Tuple2<Integer,String>, String, Integer>() {

			/**
			 * 
			 */
			private static final long serialVersionUID = 1L;

			@Override
			public Tuple2<String, Integer> call(Tuple2<Integer, String> t)
					throws Exception {
				return new Tuple2<String, Integer>(t._2, t._1);
			}
		}).take(1);
		
		System.out.println("take--------"+take);
		
		JavaPairRDD<String, Integer> result = mapToPair.filter(new Function<Tuple2<String,Integer>, Boolean>() {

			/**
			 * 
			 */
			private static final long serialVersionUID = 1L;

			@Override
			public Boolean call(Tuple2<String, Integer> v1) throws Exception {
				return !v1._1.equals(take.get(0)._1);
			}
		}).reduceByKey(new Function2<Integer,Integer,Integer>(){

			/**
			 * 
			 */
			private static final long serialVersionUID = 1L;

			@Override
			public Integer call(Integer v1, Integer v2) throws Exception {
				return v1+v2;
			}
			
		}).mapToPair(new PairFunction<Tuple2<String,Integer>, Integer, String>() {

			/**
			 * 
			 */
			private static final long serialVersionUID = 1L;

			@Override
			public Tuple2<Integer, String> call(Tuple2<String, Integer> t)
					throws Exception {
				return new Tuple2<Integer, String>(t._2, t._1);
			}
		}).sortByKey(false).mapToPair(new PairFunction<Tuple2<Integer,String>, String, Integer>() {

			/**
			 * 
			 */
			private static final long serialVersionUID = 1L;

			@Override
			public Tuple2<String, Integer> call(Tuple2<Integer, String> t)
					throws Exception {
				return new Tuple2<String, Integer>(t._2, t._1);
			}
		});
		
		result.foreach(new VoidFunction<Tuple2<String,Integer>>() {
			
			/**
			 * 
			 */
			private static final long serialVersionUID = 1L;

			@Override
			public void call(Tuple2<String, Integer> t) throws Exception {
				System.out.println(t);
			}
		});
		
		jsc.stop();
		
	}
}

3、Spark代码流程

1）、创建SparkConf对象

可以设置Application name。

可以设置运行模式。

可以设置Spark application的资源需求。

2）、创建SparkContext对象

3）、基于Spark的上下文创建一个RDD，对RDD进行处理。

4）、应用程序中要有Action类算子来触发Transformation类算子执行。

5）、关闭Spark上下文对象SparkContext。

二、Spark持久化算子

1、控制算子

1）、概念

控制算子有三种，cache，persist,checkpoint,以上算子都可以将RDD持久化，持久化单位是partition。cache和persist都是懒执行的。必须有一个action类算子触发执行。checkpoint算子不仅能将RDD持久化到磁盘，还能切断RDD之间的依赖关系。

2）、cache

默认将RDD的数据持久化到内存中。cache是懒执行。

注意：chche()=persist()=persist(StorageLevel.Memory_Only)

测试cache文件：

测试代码：

1.SparkConf conf = new SparkConf();
2.conf.setMaster("local").setAppName("CacheTest");
3.JavaSparkContext jsc = new JavaSparkContext(conf);
4.JavaRDD<String> lines = jsc.textFile("persistData.txt");
5.
6.lines = lines.cache();
7.long startTime = System.currentTimeMillis();
8.long count = lines.count();
9.long endTime = System.currentTimeMillis();
10.System.out.println("共"+count+ "条数据，"+"初始化时间+cache时间+计算时间="+ 
11.(endTime-startTime));
12.
13.long countStartTime = System.currentTimeMillis();
14.long countrResult = lines.count();
15.long countEndTime = System.currentTimeMillis();
16.System.out.println("共"+countrResult+ "条数据，"+"计算时间="+ (countEndTime-
17.countStartTime));
18.
19.jsc.stop();

persist：

可以指定持久化的级别。最常用的是MEMORY_ONLY和MEMORY_AND_DISK。”_2“表示有副本数。

持久化级别如下：

2、cache和persist的注意事项

1）、cache和persist都是懒执行，必须有一个action类算子触发执行。

2）、cache和persist算子的返回值可以赋值给一个变量，在其他job中直接使用这个变量就是使用持久化的数据了。持久化的单位是partition。

3）、cache和persist算子后不能立即紧跟action算子。

4）、cache和persist算子持久化的数据当applilcation执行完成之后会被清除。

错误：rdd.cache().count() 返回的不是持久化的RDD，而是一个数值了。

3、checkpoint

checkpoint将RDD持久化到磁盘，还可以切断RDD之间的依赖关系。checkpoint目录数据当application执行完之后不会被清除。

persist(StorageLevel.DISK_ONLY)与Checkpoint的区别？

1)、checkpoint需要指定额外的目录存储数据，checkpoint数据是由外部的存储系统管理，不是Spark框架管理，当application完成之后，不会被清空。

2）、cache() 和persist() 持久化的数据是由Spark框架管理，当application完成之后，会被清空。

3）、checkpoint多用于保存状态。

checkpoint 的执行原理：

1）、当RDD的job执行完毕后，会从finalRDD从后往前回溯。

2）、当回溯到某一个RDD调用了checkpoint方法，会对当前的RDD做一个标记。

3）、Spark框架会自动启动一个新的job，重新计算这个RDD的数据，将数据持久化到HDFS上。

优化：对RDD执行checkpoint之前，最好对这个RDD先执行cache，这样新启动的job只需要将内存中的数据拷贝到HDFS上就可以，省去了重新计算这一步。
使用：

1.SparkConf conf = new SparkConf();
2.conf.setMaster("local").setAppName("checkpoint");
3.JavaSparkContext sc = new JavaSparkContext(conf);
4.sc.setCheckpointDir("./checkpoint");
5.JavaRDD<Integer> parallelize = sc.parallelize(Arrays.asList(1,2,3));
6.parallelize.checkpoint();
7.parallelize.count();
8.sc.stop();