。

当某个RDD执行map方法时#xff0c;会遍历该RDD中的每一个数据项#xff0c;并依次应用f函…

Transformation转换算子

参数f是一个函数可以写作匿名子类它可以接收一个参数。

当某个RDD执行map方法时会遍历该RDD中的每一个数据项并依次应用f函数从而产生一个新的RDD。

即这个新RDD中的每一个元素都是原来RDD中每一个元素依次应用f函数而得到的。

1具体实现

org.apache.spark.api.java.JavaRDD;

import

org.apache.spark.api.java.JavaSparkContext;

import

org.apache.spark.api.java.function.Function;

public

SparkConf().setMaster(local[*]).setAppName(sparkCore);

创建sparkContext

mapRDD1.collect().forEach(System.out::println);

关闭sc

与map操作类似将RDD中的每一个元素通过应用f函数依次转换为新的元素并封装到RDD中。

区别在flatMap操作中f函数的返回值是一个集合并且会将每一个该集合中的元素拆分出来放到新的RDD中。

2需求说明创建一个集合集合里面存储的还是子集合把所有子集合中数据取出放入到一个大的集合中。

4具体实现

org.apache.commons.collections.ListUtils;

import

org.apache.spark.api.java.JavaRDD;

import

org.apache.spark.api.java.JavaSparkContext;

import

org.apache.spark.api.java.function.FlatMapFunction;

import

SparkConf().setMaster(local[*]).setAppName(sparkCore);

创建sparkContext

arrayLists.add(Arrays.asList(1,2,3));

arrayLists.add(Arrays.asList(4,5,6));

JavaRDDListString

collect().forEach(System.out::println);

JavaRDDString

collect().forEach(System.out::println);

关闭sc

1功能说明分组按照传入函数的返回值进行分组。

将相同的key对应的值放入一个迭代器。

3具体实现

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.Function;

import

SparkConf().setMaster(local[*]).setAppName(sparkCore);

创建sparkContext

sc.parallelize(Arrays.asList(1,

4),2);

groupByRDD.collect().forEach(System.out::println);

JavaPairRDDBoolean,

collect().forEach(System.out::println);

关闭sc

groupBy会存在shuffle过程shuffle将不同的分区数据进行打乱重组的过程shuffle一定会落盘。

可以在local模式下执行程序通过4040看效果。

1.1.4

接收一个返回值为布尔类型的函数作为参数。

当某个RDD调用filter方法时会对该RDD中每一个元素应用f函数如果返回值类型为true则该元素会被添加到新的RDD中。

3代码实现

org.apache.spark.api.java.JavaRDD;

import

org.apache.spark.api.java.JavaSparkContext;

import

org.apache.spark.api.java.function.Function;

import

SparkConf().setMaster(local[*]).setAppName(sparkCore);

创建sparkContext

sc.parallelize(Arrays.asList(1,

4),2);

collect().forEach(System.out::println);

关闭sc

org.apache.spark.api.java.JavaRDD;

import

org.apache.spark.api.java.JavaSparkContext;

import

SparkConf().setMaster(local[*]).setAppName(sparkCore);

创建sparkContext

sc.parallelize(Arrays.asList(1,

6),

collect().forEach(System.out::println);

关闭sc

该操作用于排序数据。

在排序之前可以将数据通过f函数进行处理之后按照f函数处理的结果进行排序默认为正序排列。

排序后新产生的RDD的分区数与原RDD的分区数一致。

Spark的排序结果是全局有序。

3代码实现

org.apache.spark.api.java.JavaRDD;

import

org.apache.spark.api.java.JavaSparkContext;

import

org.apache.spark.api.java.function.Function;

import

SparkConf().setMaster(local[*]).setAppName(sparkCore);

创建sparkContext

sc.parallelize(Arrays.asList(5,

11,

collect().forEach(System.out::println);

关闭sc

要想使用Key-Value类型的算子首先需要使用特定的方法转换为PairRDD

package

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.PairFunction;

import

SparkConf().setMaster(local[*]).setAppName(sparkCore);

创建sparkContext

sc.parallelize(Arrays.asList(1,

4),2);

collect().forEach(System.out::println);

关闭sc

2需求说明创建一个pairRDD并将value添加字符串|||

4代码实现

org.apache.spark.api.java.JavaPairRDD;

import

org.apache.spark.api.java.JavaSparkContext;

import

org.apache.spark.api.java.function.Function;

import

SparkConf().setMaster(local[*]).setAppName(sparkCore);

创建sparkContext

sc.parallelizePairs(Arrays.asList(new

Tuple2(k,

collect().forEach(System.out::println);

关闭sc

groupByKey对每个key进行操作但只生成一个seq并不进行聚合。

该操作可以指定分区器或者分区数默认使用HashPartitioner

2需求说明统计单词出现次数

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.Function;

import

org.apache.spark.api.java.function.PairFunction;

import

SparkConf().setMaster(local[*]).setAppName(sparkCore);

创建sparkContext

sc.parallelize(Arrays.asList(hi,hi,hello,spark

),2);

collect().forEach(System.out::println);

关闭sc

1功能说明该操作可以将RDD[K,V]中的元素按照相同的K对V进行聚合。

其存在多种重载形式还可以设置新RDD的分区数。

2需求说明统计单词出现次数

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.Function;

import

org.apache.spark.api.java.function.Function2;

import

org.apache.spark.api.java.function.PairFunction;

import

SparkConf().setMaster(local[*]).setAppName(sparkCore);

创建sparkContext

sc.parallelize(Arrays.asList(hi,hi,hello,spark

),2);

collect().forEach(System.out::println);

关闭sc

1reduceByKey按照key进行聚合在shuffle之前有combine预聚合操作返回结果是RDD[K,V]。

2groupByKey按照key进行分组直接进行shuffle。

3开发指导在不影响业务逻辑的前提下优先选用reduceByKey。

求和操作不影响业务逻辑求平均值影响业务逻辑。

影响业务逻辑时建议先对数据类型进行转换再合并。

package

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.Function;

import

org.apache.spark.api.java.function.Function2;

import

org.apache.spark.api.java.function.PairFunction;

import

SparkConf().setMaster(local[*]).setAppName(sparkCore);

创建sparkContext

sc.parallelizePairs(Arrays.asList(new

Tuple2(hi,

tuple2JavaPairRDD.reduceByKey(new

Integer,

collect().forEach(System.out::println);

关闭sc

在一个(K,V)的RDD上调用K必须实现Ordered接口返回一个按照key进行排序的(K,V)的RDD。

2需求说明创建一个pairRDD按照key的正序和倒序进行排序

3代码实现

org.apache.spark.api.java.JavaPairRDD;

import

org.apache.spark.api.java.JavaSparkContext;

import

SparkConf().setMaster(local[*]).setAppName(sparkCore);

创建sparkContext

sc.parallelizePairs(Arrays.asList(new

Tuple2(4,

collect().forEach(System.out::println);

关闭sc

行动算子是触发了整个作业的执行。

因为转换算子都是懒加载并不会立即执行。

2.1

1功能说明在驱动程序中以数组Array的形式返回数据集的所有元素。

2需求说明创建一个RDD并将RDD内容收集到Driver端打印

package

org.apache.spark.api.java.JavaRDD;

import

org.apache.spark.api.java.JavaSparkContext;

import

SparkConf().setMaster(local[*]).setAppName(sparkCore);

创建sparkContext

sc.parallelize(Arrays.asList(1,

4),2);

org.apache.spark.api.java.JavaRDD;

import

org.apache.spark.api.java.JavaSparkContext;

import

SparkConf().setMaster(local[*]).setAppName(sparkCore);

创建sparkContext

sc.parallelize(Arrays.asList(1,

4),2);

org.apache.spark.api.java.JavaRDD;

import

org.apache.spark.api.java.JavaSparkContext;

import

SparkConf().setMaster(local[*]).setAppName(sparkCore);

创建sparkContext

sc.parallelize(Arrays.asList(1,

4),2);

org.apache.spark.api.java.JavaRDD;

import

org.apache.spark.api.java.JavaSparkContext;

import

SparkConf().setMaster(local[*]).setAppName(sparkCore);

创建sparkContext

sc.parallelize(Arrays.asList(1,

4),2);

list.forEach(System.out::println);

关闭sc

org.apache.spark.api.java.JavaPairRDD;

import

org.apache.spark.api.java.JavaSparkContext;

import

SparkConf().setMaster(local[*]).setAppName(sparkCore);

创建sparkContext

sc.parallelizePairs(Arrays.asList(new

Tuple2(a,

功能说明将数据集的元素以textfile的形式保存到HDFS文件系统或者其他支持的文件系统对于每个元素Spark将会调用toString方法将它装换为文件中的文本

序列化成对象保存到文件

org.apache.spark.api.java.JavaPairRDD;

import

org.apache.spark.api.java.JavaRDD;

import

org.apache.spark.api.java.JavaSparkContext;

import

SparkConf().setMaster(local[*]).setAppName(sparkCore);

创建sparkContext

sc.parallelize(Arrays.asList(1,

4),2);

integerJavaRDD.saveAsTextFile(output);

integerJavaRDD.saveAsObjectFile(output1);

关闭sc

org.apache.spark.api.java.JavaRDD;

import

org.apache.spark.api.java.JavaSparkContext;

import

org.apache.spark.api.java.function.VoidFunction;

import

SparkConf().setMaster(local[*]).setAppName(sparkCore);

创建sparkContext

sc.parallelize(Arrays.asList(1,

4),4);

org.apache.spark.api.java.JavaRDD;

import

org.apache.spark.api.java.JavaSparkContext;

import

org.apache.spark.api.java.function.VoidFunction;

import

SparkConf().setAppName(core).setMaster(local[*]);

创建sc环境

sc.parallelize(Arrays.asList(1,

6),

parallelize.foreachPartition(new

Override