Spark学习---2、SparkCore（RDD概述、RDD编程（创建、分区规则、转换算子、Action算子））（二）-阿里云开发者社区

2.3.1.4 groupBy()分组

1、用法：groupBy(f) ,以元素为粒度对每个元素执行函数f。

2、函数f：

（1）函数f为用户自定义实现内容，返回值任意

（2）函数返回值为算子groupBy返回值的key，元素为value。

（3）算子groupBy的返回值为新的重新分区的K—V类型RDD

3、功能说明：分组，按照传入函数的返回值进行分组。将相同的key对应的值放入一个迭代器。

4、案例说明：创建一个RDD，按照元素模以2的值进行分组。

5、代码实现

package com.zhm.spark.operator.value;
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.Function;
import java.util.Arrays;
/**
 * @ClassName StudyGroupBy
 * @Description TODO
 * @Author Zouhuiming
 * @Date 2023/6/27 14:32
 * @Version 1.0
 */
public class StudyGroupBy {
    public static void main(String[] args) throws InterruptedException {
        //1、创建配置对象
        SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("sparkCore");
        //2、创建sparkContext
        JavaSparkContext sparkContext = new JavaSparkContext(conf);
        //3、Todo 根据集合创建RDD
        JavaRDD<Integer> javaRDD = sparkContext.parallelize(Arrays.asList(1, 2, 3, 4, 5), 2);
        //4、储存到文件查看groupby之前的情况
        javaRDD.saveAsTextFile("outputGroupByBefore");
        //5、对RDD执行groupBy操作，计算规则是value%2
        JavaPairRDD<Integer, Iterable<Integer>> integerIterableJavaPairRDD = javaRDD.groupBy(new Function<Integer, Integer>() {
            @Override
            public Integer call(Integer integer) throws Exception {
                return integer % 2;
            }
        });
        //6、类型可以容易修改
        JavaPairRDD<Boolean, Iterable<Integer>> booleanIterableJavaPairRDD = javaRDD.groupBy(new Function<Integer, Boolean>() {
            @Override
            public Boolean call(Integer integer) throws Exception {
                return integer % 2 == 0;
            }
        });
        //7、输出结果
        System.out.println("---执行groupBy之后的RDD分区情况---");
        integerIterableJavaPairRDD.collect().forEach(System.out::println);
        booleanIterableJavaPairRDD.collect().forEach(System.out::println);
        //x 关闭 sparkContext
        Thread.sleep(1000000000L);
        sparkContext.stop();
    }
}

运行结果：

6、说明：

（1）groupBy会存在shuffle过程

（2）shuffle：将不同的分区数据进行打乱重组的过程

（3）shuffle一定会落盘。可以在local模式下执行程序，通过4040看效果。

2.3.1.5 distinct()去重

1、用法：distinct(numPartitions), 实现对RDD进行分布式去重。

2、参数numPartitions：指定去重后的RDD的分区个数。

3、功能说明：对内部的元素去重，并将去重后的元素放到新的RDD中。

4、代码实现

package com.zhm.spark.operator.value;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import java.util.Arrays;
/**
 * @ClassName StudyDistinct
 * @Description 去重
 * @Author Zouhuiming
 * @Date 2023/6/27 14:44
 * @Version 1.0
 */
public class StudyDistinct {
    public static void main(String[] args) throws InterruptedException {
        //1、创建配置对象
        SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("sparkCore");
        //2、创建sparkContext
        JavaSparkContext sparkContext = new JavaSparkContext(conf);
        //3、Todo 根据集合创建RDD
        JavaRDD<Integer> javaRDD = sparkContext.parallelize(Arrays.asList(1, 2, 3, 1, 2, 3, 4, 5, 6), 2);
        //4、使用distinct算子实现去重，底层使用分布式去重，慢但是不会OOM
        JavaRDD<Integer> distinctRDD = javaRDD.distinct();
        //5、收集打印
        distinctRDD.collect().forEach(System.out::println);
        Thread.sleep(100000000L);
        //6 关闭 sparkContext
        sparkContext.stop();
    }
}

运行结果：

5、distinct会存在shuffle过程。

2.3.1.6 sortBy()排序

1、用法：RDD.sortBy(f, ascending, numpartitions)

2、参数介绍：

（1）函数f：对每个元素都执行函数f，返回值类型和元素中的类型一致

（2）ascending：数据类型为Boolean，默认是True，参数决定了排序后，RDD中的元素的排列顺序，即升序/降序

（3）numpartitions：排序后的RDD的分区数。

3、功能说明

该操作用于排序数据。在排序之前，可以将数据通过f函数进行处理，之后按照f函数处理的结果进行排序，默认为正序排列。默认排序后新产生的RDD的分区数与原RDD的分区数一致。Spark的排序结果是全局有序。

4、案例需求说明：创建一个RDD，按照数字大小分别实现正序和倒序排序

5、代码实现

package com.zhm.spark.operator.value;
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;
import java.util.Arrays;
/**
 * @ClassName StudySortBy
 * @Description TODO
 * @Author Zouhuiming
 * @Date 2023/6/27 14:48
 * @Version 1.0
 */
public class StudySortBy {
    public static void main(String[] args) {
        //1、创建配置对象
        SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("sparkCore");
        //2、创建sparkContext
        JavaSparkContext sparkContext = new JavaSparkContext(conf);
        //3、Todo
        JavaRDD<Integer> javaRDD = sparkContext.parallelize(Arrays.asList(1, 3, 2, 9, 6, 5, 3), 2);
        //4、使用sortBy算子对javaRDD进行排序（泛型->以谁作为标准排序，true->为正序）
        JavaRDD<Integer> javaRDD1 = javaRDD.sortBy(new Function<Integer, Integer>() {
            @Override
            public Integer call(Integer integer) throws Exception {
                return integer;
            }
        }, true, 2);
        //5、收集输出
        javaRDD1.collect().forEach(System.out::println);
        //x 关闭 sparkContext
        sparkContext.stop();
    }
}

运行结果：

2.3.2 Key-Value类型

创建包名：com.zhm.spark.operator.keyvalue

2.3.2.1 mapToPair()

1、用法：RDD.mapToPair(f) ，对父RDD中的每条记录都执行函数f，得到新的记录<k, v>

2、作用：将Value类型转换为key-Value类型

3、代码实现

package com.zhm.spark.operator.keyvalue;
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.PairFunction;
import scala.Tuple2;
import java.util.Arrays;
import java.util.Iterator;
/**
 * @ClassName StudyMapToPair
 * @Description 将不是kv类型的RDD转换为kv类型的RDD
 * @Author Zouhuiming
 * @Date 2023/6/27 14:52
 * @Version 1.0
 */
public class StudyMapToPair {
    public static void main(String[] args) {
        //1、创建配置对象
        SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("sparkCore");
        //2、创建sparkContext
        JavaSparkContext sparkContext = new JavaSparkContext(conf);
        //3、Todo
        JavaRDD<Integer> javaRDD = sparkContext.parallelize(Arrays.asList(1, 2, 3, 4, 5, 6), 2);
        //4、使用mapToPair算子对javaRDD转换为kv类型的RDD
        JavaPairRDD<Integer, Integer> pairRDD = javaRDD.mapToPair(new PairFunction<Integer, Integer, Integer>() {
            @Override
            public Tuple2<Integer, Integer> call(Integer integer) throws Exception {
                return new Tuple2<>(integer, integer);
            }
        });
        //5、收集输出
        System.out.println("------由v型RDD转换得到的kv型RDD------");
        pairRDD.collect().forEach(System.out::println);
        //Todo 由集合直接创建KV型RDD
        JavaPairRDD<Integer, Integer> integerIntegerJavaPairRDD = sparkContext.parallelizePairs(Arrays.asList(new Tuple2<>(1, 1), new Tuple2<>(2, 2), new Tuple2<>(3, 3), new Tuple2<>(4, 5)));
        //6、收集输出
        System.out.println("-----由集合直接创建KV型RDD-----");
        integerIntegerJavaPairRDD.collect().forEach(System.out::println);
     sparkContext.stop();
    }
}

运行结果：

2.3.2.2 mapValues()只对Value进行操作

1、用法：newRDD = oldRdd.mapValues(func)

2、参数函数func：自定义实现的函数，仅对oldRdd中（k,v）数据的v作用。

3、功能说明：针对于(K,V)形式的类型只对V进行操作

4、需求说明：创建一个pairRDD，并将value添加字符尾缀" Fighting"

5、代码实现

package com.zhm.spark.operator.keyvalue;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import scala.Tuple2;
import java.util.Arrays;
/**
 * @ClassName StudyMapValues
 * @Description 对kv类型的RDD的v进行操作
 * @Author Zouhuiming
 * @Date 2023/6/27 15:01
 * @Version 1.0
 */
public class StudyMapValues {
    public static void main(String[] args) {
        //1、创建配置对象
        SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("sparkCore");
        //2、创建sparkContext
        JavaSparkContext sparkContext = new JavaSparkContext(conf);
        //3、Todo
        JavaPairRDD<Integer, Integer> javaPairRDD = sparkContext.parallelizePairs(Arrays.asList(new Tuple2<>(1, 1), new Tuple2<>(2, 2), new Tuple2<>(3, 3),
                new Tuple2<>(4, 4), new Tuple2<>(5, 5)), 2);
        //4、为kv型RDD的v拼接尾缀"Fighting"
        JavaPairRDD<Integer, String> resultRDD = javaPairRDD.mapValues(new Function<Integer, String>() {
            @Override
            public String call(Integer integer) throws Exception {
                return integer + " Fighting";
            }
        });
        //5、打印收集
        resultRDD.collect().forEach(System.out::println);
        //6 关闭 sparkContext
        sparkContext.stop();
    }
}

运行结果：

2.3.2.3 groupByKey()按照K重新分组

1、用法：KVRDD.groupByKey();

2、功能说明

groupByKey对每个key进行操作，但只生成一个结果集，并不进行聚合。

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

3、需求说明

统计单词出现次数

4、代码实现

package com.zhm.spark.operator.keyvalue;
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.PairFunction;
import scala.Tuple2;
import java.util.Arrays;
/**
 * @ClassName StudyGroupByKey
 * @Description 分组聚合
 * @Author Zouhuiming
 * @Date 2023/6/27 15:07
 * @Version 1.0
 */
public class StudyGroupByKey {
    public static void main(String[] args) {
        //1、创建配置对象
        SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("sparkCore");
        //2、创建sparkContext
        JavaSparkContext sparkContext = new JavaSparkContext(conf);
        //3、Todo
        JavaRDD<String> javaRDD = sparkContext.parallelize(Arrays.asList("a", "a", "a", "b", "b", "b", "b", "a"), 2);
        //4、根据JavaRDD创建KVRDD
        JavaPairRDD<String, Integer> pairRDD = javaRDD.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String s) throws Exception {
                return new Tuple2<>(s, 1);
            }
        });
        //5、聚合相同的key
        JavaPairRDD<String, Iterable<Integer>> groupByKeyRDD = pairRDD.groupByKey();
        //6、收集并输出RDD内容
        System.out.println("-----查看groupByKeyRDD的内容------");
        groupByKeyRDD.collect().forEach(System.out::println);
        sparkContext.stop();
    }
}

5、运行结果：

2.3.2.4 reduceByKey()按照K聚合V

1、用法：KVRDD.reduceByKey(f);

2、功能说明：将RDD[K,V]中的元素按照相同的K的V进行聚合。存在多种重载形式，可设置新RDD的分区数。

3、需求说明：统计单词出现次数

4、代码实现

package com.zhm.spark.operator.keyvalue;
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.Function2;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;
import java.util.Arrays;
/**
 * @ClassName StudyReduceByKey
 * @Description TODO
 * @Author Zouhuiming
 * @Date 2023/6/27 15:15
 * @Version 1.0
 */
public class StudyReduceByKey {
    public static void main(String[] args) {
        //1、创建配置对象
        SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("sparkCore");
        //2、创建sparkContext
        JavaSparkContext sparkContext = new JavaSparkContext(conf);
        //3、Todo
        JavaRDD<String> javaRDD = sparkContext.parallelize(Arrays.asList("a", "a", "a", "b", "b", "b", "b", "a"), 2);
        //4、根据JavaRDD创建KVRDD
        JavaPairRDD<String, Integer> javaPairRDD = javaRDD.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String s) throws Exception {
                return new Tuple2<>(s, 1);
            }
        });
        //5、聚合相同的key，统计单词出现的次数
        JavaPairRDD<String, Integer> result = javaPairRDD.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer integer, Integer integer2) throws Exception {
                return integer + integer2;
            }
        });
        //6、收集并输出RDD的内容
        System.out.println("查看--result--的内容");
        result.collect().forEach(System.out::println);
        //x 关闭 sparkContext
        sparkContext.stop();
    }
}

运行结果：

2.3.2.5 reduceByKey和groupByKey区别

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

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

3、在不影响业务逻辑的前提下，优先选用reduceByKey。求和操作不影响业务逻辑，求平均值影响业务逻辑。影响业务逻辑时建议先对数据类型进行转换再合并。

2.3.2.6 sortByKey()按照K进行排序

1、用法：kvRDD.sortByKey(true/false)

2、功能说明：在一个(K,V)的RDD上调用，K必须实现Ordered接口，返回一个按照key进行排序的(K,V)的RDD。

3、参数说明：true:为升序排列，false为降序排列

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

5、代码实现

package com.zhm.spark.operator.keyvalue;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;
import java.util.Arrays;
/**
 * @ClassName StudySortByKey
 * @Description TODO
 * @Author Zouhuiming
 * @Date 2023/6/27 15:47
 * @Version 1.0
 */
public class StudySortByKey {
    public static void main(String[] args) {
        //1、创建配置对象
        SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("sparkCore");
        //2、创建sparkContext
        JavaSparkContext sparkContext = new JavaSparkContext(conf);
        //3、Todo 创建kvRDD
        JavaPairRDD<Integer, String> javaPairRDD = sparkContext.parallelizePairs(Arrays.asList(new Tuple2<>(4, "d"), new Tuple2<>(3, "c"), new Tuple2<>(1, "a"),
                new Tuple2<>(2, "b")));
        //4、收集输出
        System.out.println("排序前：");
        javaPairRDD.collect().forEach(System.out::println);
        //5、对RDD按照key进行排序
        JavaPairRDD<Integer, String> sortByKeyRDD = javaPairRDD.sortByKey(true);
        //收集输出
        System.out.println("排序后：");
        sortByKeyRDD.collect().forEach(System.out::println);
        //x 关闭 sparkContext
        sparkContext.stop();
    }
}

运行结果:

2.4 Action行动算子

行动算子是触发了整个作业的执行。因为转换算子都是懒加载，并不会立即执行。

创建包名：com.zhm.spark.operator.action

2.4.1 collect():以数组的形式返回数据集

1、用法：RDD.collect();

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

注意：所以的数据都会被拉取到Driver端，慎用

3、需求说明：创建一个RDD，并将RDD内容收集到Driver端打印（代码实现在最后面）

2.4.2 count()返回RDD中元素个数

1、用法：RDD.count(),返回值为Long类型

2、功能说明：返回RDD中元素的个数

3、需求说明：创建一个RDD，统计该RDD的条数（代码实现在最后面）

2.4.3 first()返回RDD中的第一个元素

1、用法：RDD.first(), 返回值类型是元素类型

2、功能说明：返回RDD中的第一个元素

3、需求说明：创建一个RDD，返回该RDD中的第一个元素（代码实现在最后面）

2.4.4 take()返回由RDD前n个元素组成的数组

1、RDD.take(int num), 返回值为RDD中元素类型的List列表

2、功能说明：返回一个由RDD的前n个元素组成的数组

3、需求说明：创建一个RDD，取出前3个元素（代码实现在最后面）

2.4.5 countByKey()统计每种key的个数

1、用法：pairRDD.countByKey()，返回值类型为Map<[RDD中key的类型] , Long>

2、功能说明：统计每种key的个数

3、需求说明：创建一个PairRDD，统计每种key的个数（代码实现在最后面）

2.4.6 save相关算子

1、saveAsTextFile(path)

（1）功能：将RDD保存成Text文件

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

2、 saveAsObjectFile(path)

（1）功能：序列化成对象保存到文件

（2）功能说明：用于将RDD中的元素序列化成对象，存储到文件中。

（代码实现在最后面）

2.4.7 foreach()遍历RDD中每一个元素

1、功能：遍历RDD中的每个元素，并依次应用函数

2、需求说明：创建一个RDD，对每个元素进行打印

2.4.8 所有的代码实现

package com.zhm.spark.operator.action;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;
import java.util.Arrays;
/**
 * @ClassName TestAll
 * @Description TODO
 * @Author Zouhuiming
 * @Date 2023/6/28 13:47
 * @Version 1.0
 */
public class TestAll {
    public static void main(String[] args) {
        //设置往HDFS储存数据的用户名
        System.setProperty("HADOOP_USER_NAME","zhm");
        //1、创建配置对象
        SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("TestAll");
        //2、创建sparkContext
        JavaSparkContext sparkContext = new JavaSparkContext(conf);
        //3、获取RDD
//        JavaRDD<Integer> javaRDD = sparkContext.parallelize(Arrays.asList(1, 2, 3, 4, 5, 6, 7));
        JavaPairRDD<Integer, String> javaPairRDD = sparkContext.parallelizePairs(Arrays.asList(new Tuple2<>(1, "z"), new Tuple2<>(2, "h"), new Tuple2<>(3, "m"),
                new Tuple2<>(1, "zhm")
        ),2);
        //4、collect():以数组的形式返回数据集
        //注意：所有的数据都会被拉取到Driver端，慎用
        System.out.println("-------------collect测试-------------");
        javaPairRDD.collect().forEach(System.out::println);
        //5、count():返回RDD中元素个数
        System.out.println("-------------count测试-------------");
        System.out.println(javaPairRDD.collect());
        //6、first():返回RDD中第一个元素
        System.out.println("-------------first测试-------------");
        System.out.println(javaPairRDD.first());
        //7、take():返回RDD前n个元素组成的数组
        System.out.println("-------------take测试-------------");
        javaPairRDD.take(3).forEach(System.out::println);
        //8、countByKey统计每种key的个数
        System.out.println("-------------countByKey测试-------------");
        System.out.println(javaPairRDD.countByKey());
        //9、save相关的算子
        //以文本格式储存数据
        javaPairRDD.saveAsTextFile("outputText");
        //以对象储存数据
        javaPairRDD.saveAsObjectFile("outputObject");
        //10、foreach()：遍历RDD中每一个元素
        javaPairRDD.foreach(new VoidFunction<Tuple2<Integer, String>>() {
            @Override
            public void call(Tuple2<Integer, String> integerStringTuple2) throws Exception {
                System.out.println(integerStringTuple2._1+":"+integerStringTuple2._2);
            }
        });
        //x 关闭 sparkContext
        sparkContext.stop();
    }
}

运行结果

Spark学习---2、SparkCore（RDD概述、RDD编程（创建、分区规则、转换算子、Action算子））（二）

2.3.1.4 groupBy()分组

2.3.1.6 sortBy()排序

2.3.2 Key-Value类型

2.4 Action行动算子

2.4.1 collect():以数组的形式返回数据集

2.4.3 first()返回RDD中的第一个元素

2.4.4 take()返回由RDD前n个元素组成的数组

2.4.5 countByKey()统计每种key的个数

2.4.6 save相关算子

2.4.7 foreach()遍历RDD中每一个元素

2.4.8 所有的代码实现

热门文章

最新文章

相关课程

相关电子书