做房产买卖哪些网站可以获客,php 网站模板 x11,沐浴露营销软文,北京官方网站怎么做一、转换算子和行动算子
1、Transformations转换算子
1#xff09;、概念
Transformations类算子是一类算子#xff08;函数#xff09;叫做转换算子#xff0c;如map、flatMap、reduceByKey等。Transformations算子是延迟执行#xff0c;也叫懒加载执行。
2)、Transf…一、转换算子和行动算子
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:作用在KV格式的RDD上对Key进行升序或者降序排序。
2、Action行动算子
1、概念
Action类算子也是一类算子函数叫做行动算子如foreach,collect,count等。Transformations类算子是延迟执行Action类算子是触发执行。一个application应用程序中有几个Action类算子执行就有几个job运行。
2、Action类算子
count返回数据集中的元素数。会在结果计算完成后回收到Driver端。
take(n):返回一个包含数据集前n个元素的集合。
firstfirsttake(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);JavaRDDString lines jsc.textFile(./records.txt);JavaRDDString flatMap lines.flatMap(new FlatMapFunctionString, String() {/*** */private static final long serialVersionUID 1L;Overridepublic IterableString call(String t) throws Exception {return Arrays.asList(t.split( ));}});JavaPairRDDString, Integer mapToPair flatMap.mapToPair(new PairFunctionString,String, Integer() {/*** */private static final long serialVersionUID 1L;Overridepublic Tuple2String, Integer call(String t) throws Exception {return new Tuple2String, Integer(t, 1);}});JavaPairRDDString, Integer sample mapToPair.sample(true, 0.5);final ListTuple2String, Integer take sample.reduceByKey(new Function2Integer,Integer,Integer(){/*** */private static final long serialVersionUID 1L;Overridepublic Integer call(Integer v1, Integer v2) throws Exception {return v1v2;}}).mapToPair(new PairFunctionTuple2String,Integer, Integer, String() {/*** */private static final long serialVersionUID 1L;Overridepublic Tuple2Integer, String call(Tuple2String, Integer t)throws Exception {return new Tuple2Integer, String(t._2, t._1);}}).sortByKey(false).mapToPair(new PairFunctionTuple2Integer,String, String, Integer() {/*** */private static final long serialVersionUID 1L;Overridepublic Tuple2String, Integer call(Tuple2Integer, String t)throws Exception {return new Tuple2String, Integer(t._2, t._1);}}).take(1);System.out.println(take--------take);JavaPairRDDString, Integer result mapToPair.filter(new FunctionTuple2String,Integer, Boolean() {/*** */private static final long serialVersionUID 1L;Overridepublic Boolean call(Tuple2String, Integer v1) throws Exception {return !v1._1.equals(take.get(0)._1);}}).reduceByKey(new Function2Integer,Integer,Integer(){/*** */private static final long serialVersionUID 1L;Overridepublic Integer call(Integer v1, Integer v2) throws Exception {return v1v2;}}).mapToPair(new PairFunctionTuple2String,Integer, Integer, String() {/*** */private static final long serialVersionUID 1L;Overridepublic Tuple2Integer, String call(Tuple2String, Integer t)throws Exception {return new Tuple2Integer, String(t._2, t._1);}}).sortByKey(false).mapToPair(new PairFunctionTuple2Integer,String, String, Integer() {/*** */private static final long serialVersionUID 1L;Overridepublic Tuple2String, Integer call(Tuple2Integer, String t)throws Exception {return new Tuple2String, Integer(t._2, t._1);}});result.foreach(new VoidFunctionTuple2String,Integer() {/*** */private static final long serialVersionUID 1L;Overridepublic void call(Tuple2String, 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、概念
控制算子有三种cachepersist,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.JavaRDDString 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.JavaRDDInteger parallelize sc.parallelize(Arrays.asList(1,2,3));
6.parallelize.checkpoint();
7.parallelize.count();
8.sc.stop();
