【spark-spring boot】学习笔记

说明

本文依赖于上一篇文章：【spark学习】 spring boot 整合 spark 的相关内容，请先阅读上一篇文章，将spark需要的环境先配置好，并测试通过之后再进行此文章的阅读和操作。

RDD学习

RDD介绍

想象一下你有一个大大的数据表，里面包含了很多很多的信息。如果你想对这些数据进行操作，比如筛选出符合条件的数据、或者对数据做一些计算，RDD 就是 Spark 用来存储和操作这些数据的一种方式。它有两个基本操作：转换操作（就像是加工数据）和 行动操作（获取结果）

RDD案例

基于集合创建RDD

import lombok.extern.slf4j.Slf4j;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;import javax.annotation.PostConstruct;
import java.util.Arrays;
import java.util.List;/*** spark 案例*/
@Slf4j
@Component
public class DemoTimer {@AutowiredJavaSparkContext javaSparkContext;@AutowiredSparkSession sparkSession;/*** 创建 RDD 算子*/@PostConstructpublic void createRDD() {// 从集合创建 RDDList<String> lists = Arrays.asList("hello", "spark", "hi", "spark", "hadoop");JavaRDD<String> parallelize = javaSparkContext.parallelize(lists); // 创建RDDparallelize.collect().forEach(System.out::println); // 打印}
}

输出结果：

RDD存入外部文件中

将RDD存入外部文件，用于观察文件结果和分区结果。txt文件数等于默认分区数。从结果中看出默认分区为4个。

注意：存放结果的文件夹路径必须没有被创建。

以下​案例中，将基于集合生成的RDD保存到saveRddDemo文件夹中​。

package www.zhangxiaosan.top.timer;import lombok.extern.slf4j.Slf4j;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;import javax.annotation.PostConstruct;
import java.util.Arrays;
import java.util.List;/*** spark 案例*/
@Slf4j
@Component
public class DemoTimer {@AutowiredJavaSparkContext javaSparkContext;@AutowiredSparkSession sparkSession;/*** 创建 RDD 算子*/@PostConstructpublic void createRDD() {// 从集合创建 RDDList<String> lists = Arrays.asList("hello", "spark", "hi", "spark", "hadoop");// parallelize(): 创建RDD，RDD中创建默认分区数// parallelize( 元素, 分区数)： 创建RDD，RDD中指定分区数JavaRDD<String> parallelize = javaSparkContext.parallelize(lists); // 创建RDDparallelize.collect().forEach(System.out::println); // 打印// 存储的目录文件夹路径。此处为项目中的路径且目录必须为不存在的路径。String fileSavePath="I:\\zhang\\SpringBootSparkDemo\\src\\main\\resources\\saveRddDemo";parallelize.saveAsTextFile(fileSavePath);// 开始将RDD保存到文件中 }
}

​运行结果：
在文件夹中存放的是运行程序生成的文件。如下图。
_SUCCESS文件​：成功标记
part-XXX 文件：保存的数据文件 ​，结果中有4个文件，说明默认分区为4个。

转换算子 操作

转换算子（Transformation）是指那些返回一个新的 RDD 的操作。转换算子不会立即执行计算，而是构建一个执行计划，只有当行动算子（Action）触发计算时，转换算子的操作才会实际执行。转换算子可以用来处理和转换数据，比如映射、过滤、聚合等。

map 操作

说明

map() 方法传入一个函数作为参数。map() 方法会将RDD中的元素逐一进行调用，函数的参数即是RDD的元素。 如：map( 函数( RDD元素 ) )。 map() 方法会返回一个新的RDD。

案例

将RDD中的每个元素都拼接上字符串 “ say hi ”

package www.zhangxiaosan.top.timer;import lombok.extern.slf4j.Slf4j;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;import javax.annotation.PostConstruct;
import java.util.Arrays;
import java.util.List;/*** spark 案例*/
@Slf4j
@Component
public class DemoTimer {@AutowiredJavaSparkContext javaSparkContext;@AutowiredSparkSession sparkSession;/*** 创建 RDD 算子*/@PostConstructpublic void createRDD() {// 从集合创建 RDDList<String> lists = Arrays.asList("张三", "李四", "王五");JavaRDD<String> parallelize = javaSparkContext.parallelize(lists); // 创建RDD// lambda表达式传入匿名函数，进行元素拼接字符。// item 表示RDD中的元素parallelize = parallelize.map(item -> item + " say hi"); parallelize.collect().forEach(System.out::println); // 打印}
}

输出结果：

flatMap操作

说明

flatMap() 方法传入一个函数作为参数 。flatMap() 方法会将RDD中的元素逐一进行调用，函数的参数即是RDD的元素。 如：flatMap( 函数( RDD元素 ) )。 flatMap() 方法会返回一个新的RDD。
flatMap() 方法 会将 RDD的元素扁平化处理成一个集合。

例如：
假设你有一个箱子，箱子里面放着几个小盒子，每个小盒子里又有一些玩具。flatMap 就是一个工具，它能帮你把每个小盒子里的玩具拿出来，直接放进一个大盒子里，最终把所有玩具放在一个地方。
每个小盒子里的玩具可以不止一个，甚至可能没有玩具（比如有的盒子是空的）。flatMap 会把每个盒子里的玩具都拿出来，放到一个大盒子里，最终得到一个扁平的大盒子，里面是所有玩具。

案例

将集合：[ [1, 2, 3, 4, 5],[hello, spark],[张三, 李四] ]
扁平化处理成：[ 1, 2, 3, 4, 5, hello, spark, 张三, 李四]

import lombok.extern.slf4j.Slf4j;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import java.util.Arrays;
import java.util.List;/*** spark 案例*/
@Slf4j
@Component
public class DemoTimer {@AutowiredJavaSparkContext javaSparkContext;@AutowiredSparkSession sparkSession;@PostConstructpublic void flatMapDemo() {// 声明集合：[ [1, 2, 3, 4, 5],[hello, spark],[张三, 李四] ]List<List<String>> arrs = Arrays.asList(Arrays.asList("1", "2", "3", "4", "5"),Arrays.asList("hello", "spark"),Arrays.asList("张三", "李四"));//输出声明的集合呢容System.out.println("原始集合打印：");arrs.forEach(item -> System.out.print(" " + item));// 分隔符System.out.println();System.out.println("-----------");System.out.println("flatMap操作后：");// 创建集合的RDDJavaRDD<List<String>> parallelize = javaSparkContext.parallelize(arrs);// flatMap操作List<String> collect = parallelize.flatMap(i -> i.iterator()).collect(); // 打印 flatMap操作 后的集合collect.forEach(item -> System.out.print(" " + item));System.out.println();}
}

filter 操作

说明

filter() 方法传入一个函数作为参数，函数返回值只能为Boolean值。filter() 方法会将RDD中的元素逐一进行调用，函数的参数即是RDD的元素。 如：filter( 函数( RDD元素 ) )。 filter() 方法会返回一个新的RDD。filter() 将RDD元素中满足条件的元素保留，即函数返回为true的元素；RDD中不满足条件的元素，即函数返回为false的元素过滤。

案例

过滤单数，保留双数

import lombok.extern.slf4j.Slf4j;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;import javax.annotation.PostConstruct;
import java.util.Arrays;
import java.util.List;/*** spark 案例*/
@Slf4j
@Component
public class DemoTimer {@AutowiredJavaSparkContext javaSparkContext;@AutowiredSparkSession sparkSession;@PostConstructpublic void filterDemo() {// 声明集合List<Integer> arrs = Arrays.asList(-1, 0, 1, 2, 3, 4, 5);System.out.println("原始集合打印：");arrs.forEach(item -> System.out.print(" " + item));// 输出过滤前的数据System.out.println();System.out.println("-----------");System.out.println("filter操作后：");JavaRDD<Integer> parallelize = javaSparkContext.parallelize(arrs);// 过滤掉 单数 （取模不等于0的数字为单数，否则为偶数）。item % 2 == 0 的数字为双数，返回有true保留List<Integer> collect = parallelize.filter(item -> item % 2 == 0).collect();// 输出过滤后的数据collect.forEach(item -> System.out.print(" " + item));System.out.println();}
}

运行结果：

groupBy 操作

说明

将数据按某些条件进行分组。每个组由键值对组成。
groupBy() 常和以下聚合函数一起使用，来对分组数据进行统计分析。
常用的聚合操作包括：

1. count(): 统计每个组的元素数量
2. sum(): 计算每个组的元素总和
3. avg(): 计算每个组的平均值
4. max()：计算每个组的最大值
5. min(): 计算每个组的最小值
6. agg(): 自定义聚合操作，可以结合多个聚合函数

案例

给写生成绩按照分数来分组。

import lombok.extern.slf4j.Slf4j;
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.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import scala.Tuple2;
import javax.annotation.PostConstruct;
import java.util.*;/*** spark 案例*/
@Slf4j
@Component
public class DemoTimer {@AutowiredJavaSparkContext javaSparkContext;@AutowiredSparkSession sparkSession;@PostConstructpublic void groupByDemo() {// 定义学生成绩数据List<Tuple2<String, Integer>> students = Arrays.asList(new Tuple2<>("张三", 85),new Tuple2<>("李四", 90),new Tuple2<>("王五", 85),new Tuple2<>("赵六", 95),new Tuple2<>("孙七", 90));// 创建RDDJavaRDD<Tuple2<String, Integer>> parallelize = javaSparkContext.parallelize(students);// 使用 groupBy 进行分组，按成绩分组JavaPairRDD<Object, Iterable<Tuple2<String, Integer>>> integerIterableJavaPairRDD = parallelize.groupBy(tuple -> tuple._2());// 使用 groupBy 按成绩分组。_2表示元组的第二个元素，即分数// 打印结果integerIterableJavaPairRDD.sortByKey().foreach(item -> System.out.println("成绩：" + item._1() + "，学生：" + item._2()));}
}

结果如下：

distinct 操作

说明

对RDD的元素进行分布式去重。返回新的RDD，新的RDD内的元素不重复出现。

案例

将集合中重复的元素去除。

import lombok.extern.slf4j.Slf4j;
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.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import scala.Tuple2;
import javax.annotation.PostConstruct;
import java.util.*;/*** spark 案例*/
@Slf4j
@Component
public class DemoTimer {@AutowiredJavaSparkContext javaSparkContext;@AutowiredSparkSession sparkSession;@PostConstructpublic void distinctDemo() {// 声明集合List<Integer> arrs = Arrays.asList(1,2,3,4,5,6,1,4,6);// 创建RDDJavaRDD<Integer> parallelize = javaSparkContext.parallelize(arrs);// distinct()方法，用于去除重复元素。JavaRDD<Integer> distinct = parallelize.distinct();// 打印结果distinct.collect().forEach(System.out::println);}
}

运行结果：

sortBy 操作

说明

对RDD的元素进行排序，返回新的RDD。
sortBy() 可传入3个参数：
参数1：函数，每个RDD元素都会传入此函数中，此函数定义排序规则。
参数2：boolean值。定义排序顺序。true为升序，false降序。
参数3：Integer 整数，指定分区数量。

案例

将集合中的元素降序排序。

import lombok.extern.slf4j.Slf4j;
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.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import scala.Tuple2;
import javax.annotation.PostConstruct;
import java.util.*;/*** spark 案例*/
@Slf4j
@Component
public class DemoTimer {@AutowiredJavaSparkContext javaSparkContext;@AutowiredSparkSession sparkSession;@PostConstructpublic void sortByDemo() {// 创建无序集合List<Integer> arrs = Arrays.asList(7,2,5,4,1,9,6,3,8);// 创建RDDJavaRDD<Integer> parallelize = javaSparkContext.parallelize(arrs);// 使用 sortBy 进行排序，按元素值排序，并返回一个RDD。JavaRDD<Integer> sortBy = parallelize.sortBy(item -> item, false, 1);// 打印结果sortBy.collect().forEach(System.out::println);}
}

运行结果：

mapToPair 操作

说明

mapToPair()是将一个普通的 RDD 转换为 JavaPairRDD 的一个方法。 JavaPairRDD 中的每个元素都是一个键值对。
mapToPair对RDD的元素进行映射成一个由键值对组成的 RDD（即映射成JavaPairRDD），即将每个元素转换成一个 Tuple2 对象，其中第一个元素是键（key），第二个元素是值（value）。

案例

将集合 [“张三:85”, “李四:90”, “王五:85”, “赵六:95”, “孙七:90”] 中的学生和成绩按照冒号分隔，形成键值对，姓名为键，值为成绩。

import lombok.extern.slf4j.Slf4j;
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 org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import scala.Tuple2;
import javax.annotation.PostConstruct;
import java.util.*;/*** spark 案例*/
@Slf4j
@Component
public class DemoTimer {@AutowiredJavaSparkContext javaSparkContext;@AutowiredSparkSession sparkSession;@PostConstructpublic void mapToPairDemo() {// 创建包含学生成绩的列表List<String> students = Arrays.asList("张三:85", "李四:90", "王五:85", "赵六:95", "孙七:90");// 将数据并行化为 RDDJavaRDD<String> studentsRDD = javaSparkContext.parallelize(students);// 使用 mapToPair 将每个元素转换为 (姓名, 成绩) 的键值对JavaPairRDD<String, Integer> studentsPairRDD = studentsRDD.mapToPair((PairFunction<String, String, Integer>) s -> {// 根据冒号分隔学生姓名和成绩，返回一个 (姓名, 成绩) 的元组String[] parts = s.split(":");// 返回一个 (姓名, 成绩) 的元组return new Tuple2<>(parts[0], Integer.parseInt(parts[1]));});// 打印结果studentsPairRDD.collect().forEach(System.out::println);}
}

运行结果：

mapValues操作

说明

mapValues() 是一个用于处理 JavaPairRDD 的方法。它可以对 RDD 中的每个键值对的 值（value） 进行转换，同时保留原来的 键（key） 不变。

案例

将集合 [“张三:85”, “李四:90”, “王五:85”, “赵六:95”, “孙七:90”] 中的学生和成绩按照冒号分隔，形成键值对，姓名为键，值为成绩。并将成绩在原分数上+5分。

import lombok.extern.slf4j.Slf4j;
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 org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import scala.Tuple2;
import javax.annotation.PostConstruct;
import java.util.*;/*** spark 案例*/
@Slf4j
@Component
public class DemoTimer {@AutowiredJavaSparkContext javaSparkContext;@AutowiredSparkSession sparkSession;@PostConstructpublic void mapToPairDemo() {// 创建包含学生成绩的列表List<String> students = Arrays.asList("张三:85", "李四:90", "王五:85", "赵六:95", "孙七:90");// 将数据并行化为 RDDJavaRDD<String> studentsRDD = javaSparkContext.parallelize(students);// 使用 mapToPair 将每个元素转换为 (姓名, 成绩) 的键值对JavaPairRDD<String, Integer> studentsPairRDD = studentsRDD.mapToPair((PairFunction<String, String, Integer>) s -> {// 根据冒号分隔学生姓名和成绩，返回一个 (姓名, 成绩) 的元组String[] parts = s.split(":");// 返回一个 (姓名, 成绩) 的元组return new Tuple2<>(parts[0], Integer.parseInt(parts[1]));});// 打印结果System.out.println("原始分数：");studentsPairRDD.collect().forEach(System.out::println);System.out.println("\n+5分后：");// item 表示每个键值对中的值，即分数JavaPairRDD<String, Integer> newValeRdd = studentsPairRDD.mapValues(item -> item + 5);// 打印结果newValeRdd.collect().forEach(System.out::println);}
}

运行结果：

groupByKey操作

说明

groupByKey() 是一个用于处理 JavaPairRDD 的方法。它根据键对数据进行分组，将所有具有相同键的元素聚集到一起，生成一个新的 RDD，其中每个键对应一个包含所有相同键值的集合。
每个键对应的值收集到一个 Iterable 容器中，然后返回一个新的 RDD，其中每个键对应一个包含该键的所有值的 Iterable。

案例

将集合 [“张三:85”, “李四:90”, “陈八:95”, “王五:85”, “黄六:92”] 中的学生和成绩按照冒号分隔，形成键值对，成绩为键，值为姓名。并将键值对中的键进行分类，将相同分数的学生归成一组。

import lombok.extern.slf4j.Slf4j;
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 org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import scala.Tuple2;
import javax.annotation.PostConstruct;
import java.util.*;/*** spark 案例*/
@Slf4j
@Component
public class DemoTimer {@AutowiredJavaSparkContext javaSparkContext;@AutowiredSparkSession sparkSession;@PostConstructpublic void groupByKeyDemo(){// 创建包含学生成绩的列表List<String> students = Arrays.asList("张三:85", "李四:90", "陈八:95", "王五:85", "黄六:92");// 将数据并行化为 RDDJavaRDD<String> studentsRDD = javaSparkContext.parallelize(students);// 使用 mapToPair 将每个元素转换为 (成绩, 姓名) 的键值对JavaPairRDD<Integer, String> studentsPairRDD = studentsRDD.mapToPair(s -> {String[] parts = s.split(":");return new Tuple2<Integer, String>(Integer.parseInt(parts[1]), parts[0]);});// 根据键值对中的键进行分类JavaPairRDD<Integer, Iterable<String>> groupedRDD = studentsPairRDD.groupByKey();// 打印结果groupedRDD.collect().forEach(pair -> {System.out.println("成绩: " + pair._1() + ", 姓名: " + pair._2());});}
}

执行结果：

reduceByKey操作

说明

reduceByKey()用于对 JavaPairRDD 数据进行聚合的一个方法。它根据键对值进行合并，并在分区内进行局部聚合，从而减少了跨节点的数据传输，通常比 groupByKey() 更高效。

案例

计算学生的总分。

import lombok.extern.slf4j.Slf4j;
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 org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import scala.Tuple2;
import javax.annotation.PostConstruct;
import java.util.*;/*** spark 案例*/
@Slf4j
@Component
public class DemoTimer {@AutowiredJavaSparkContext javaSparkContext;@AutowiredSparkSession sparkSession;@PostConstructpublic void groupByKeyDemo(){// 创建包含学生成绩的列表List<String> students = Arrays.asList("张三:85", "李四:90", "张三:95", "王五:85", "李四:92");// 将数据并行化为 RDDJavaRDD<String> studentsRDD = javaSparkContext.parallelize(students);// 使用 mapToPair 将每个元素转换为 (姓名, 成绩) 的键值对JavaPairRDD<String, Integer> studentsPairRDD = studentsRDD.mapToPair(s -> {String[] parts = s.split(":");return new Tuple2<>(parts[0], Integer.parseInt(parts[1]));});// 使用 groupByKey 按学生姓名进行成绩求和JavaPairRDD<String, Integer> groupedRDD = studentsPairRDD.reduceByKey((a, b) -> a + b);// 打印结果groupedRDD.collect().forEach(pair -> {System.out.println("姓名: " + pair._1() + ", 成绩: " + pair._2());});}
}

运行结果：

sortByKey操作

说明

对 JavaPairRDD 数据按键进行排序的方法。默认为升序。
传入参数为Boolean值：true 升序 ，false降序

案例

将学生成绩按照成绩从高到低降序排序

import lombok.extern.slf4j.Slf4j;
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 org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import scala.Tuple2;
import javax.annotation.PostConstruct;
import java.util.*;/*** spark 案例*/
@Slf4j
@Component
public class DemoTimer {@AutowiredJavaSparkContext javaSparkContext;@AutowiredSparkSession sparkSession;@PostConstructpublic void sortByKeyDemo(){// 定义原始数据List<Tuple2<String, Integer>> students = Arrays.asList(new Tuple2<>("张三", 85),new Tuple2<>("李四", 90),new Tuple2<>("王五", 65),new Tuple2<>("赵六", 95),new Tuple2<>("孙七", 90));// 生成RDDJavaPairRDD<String, Integer> studentsPairRDD = javaSparkContext.parallelizePairs(students);// 将学生原集合（姓名，成绩）格式的数据转换为（成绩，姓名）格式的键值对JavaPairRDD<Integer, String> integerStringJavaPairRDD = studentsPairRDD.mapToPair(item -> new Tuple2<>(item._2(), item._1()));// 将（成绩，姓名）格式的键值对按照键降序排序，即按照成绩降序排序JavaPairRDD<Integer, String> stringIntegerJavaPairRDD = integerStringJavaPairRDD.sortByKey(false);// 打印结果stringIntegerJavaPairRDD .collect() .forEach(pair -> {System.out.println("成绩: " + pair._1() + ", 姓名: " + pair._2());});}
}

运行结果：

行动算子 操作

行动算子（Action）是指会触发 Spark 作业的执行，并且会产生一个结果或者副作用的操作。与 转换算子（Transformation）不同，转换算子只会定义数据转换的计算逻辑，而不会立即执行。只有在遇到行动算子时，Spark 才会真正开始计算，并将结果返回给用户或写入外部存储。

当你调用一个行动算子时，Spark 会从头开始执行所有必要的转换操作，并将结果返回给你或者存储到外部系统（如 HDFS、数据库等）。

行动算子通常会返回一个具体的结果，例如一个列表、一个数值，或者在某些情况下，可能会执行一些副作用操作（例如将数据写入磁盘）。

collect 操作

说明

将分布式 RDD 中的所有数据项拉取到本地驱动程序（Driver）中，通常作为一个数组、列表或其他集合类型。因为 collect() 会将整个 RDD 数据集拉到本地，所以如果数据量非常大，可能会导致内存溢出（OutOfMemoryError）。

注意事项：
数据量大时的风险：如果 RDD 中包含的数据量非常大，调用 collect() 会导致所有数据被加载到本地驱动程序的内存中，这可能会导致内存溢出错误（OutOfMemoryError）。因此，建议在数据集非常大的情况下谨慎使用 collect()。

建议：对于大规模数据集，通常会使用其他行动算子（如 take()）获取一个数据的子集，避免一次性加载所有数据。

并行计算的代价：尽管 collect() 会将所有数据从分布式环境中拉回到单个节点，但它不会对数据进行额外的计算，只会执行之前定义的转换操作。因此，它本质上是将整个数据集的计算结果从集群中汇总回来。

案例

收集RDD数据并打印。

import lombok.extern.slf4j.Slf4j;
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 org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import scala.Tuple2;
import javax.annotation.PostConstruct;
import java.util.*;/*** spark 案例*/
@Slf4j
@Component
public class DemoTimer {@AutowiredJavaSparkContext javaSparkContext;@AutowiredSparkSession sparkSession;@PostConstructpublic void collectDemo(){// 定义一个整数集合List<Integer> arrs = Arrays.asList(1, 2, 3, 4, 5);// 创建RDDJavaRDD<Integer> parallelize = javaSparkContext.parallelize(arrs);// collect()方法，用于将RDD中的数据收集到Driver端，并返回一个List。List<Integer> collect = parallelize.collect();// 打印结果collect.forEach(item -> System.out.println(item));}
}

执行结果：

count 操作

说明

统计RDD中的元素数量。

案例

统计RDD中的元素数量，并输出

import lombok.extern.slf4j.Slf4j;
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 org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import scala.Tuple2;
import javax.annotation.PostConstruct;
import java.util.*;/*** spark 案例*/
@Slf4j
@Component
public class DemoTimer {@AutowiredJavaSparkContext javaSparkContext;@AutowiredSparkSession sparkSession;@PostConstructpublic void countDemo(){// 定义一个整数集合List<Integer> arrs = Arrays.asList(1, 2, 3, 4, 5);// 创建RDDJavaRDD<Integer> parallelize = javaSparkContext.parallelize(arrs);// count()方法，统计数量。Long total = parallelize.count();// 打印结果System.out.println("元素数量 = " + total);}
}

运行结果：

first操作

说明

返回RDD中的第一个元素

案例

获取RDD中的第一个元素并打印

import lombok.extern.slf4j.Slf4j;
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 org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import scala.Tuple2;
import javax.annotation.PostConstruct;
import java.util.*;/*** spark 案例*/
@Slf4j
@Component
public class DemoTimer {@AutowiredJavaSparkContext javaSparkContext;@AutowiredSparkSession sparkSession;@PostConstructpublic void firstDemo(){// 创建一个集合List<Integer> arrs = Arrays.asList(1, 2, 3, 4, 5);// 创建RDDJavaRDD<Integer> parallelize = javaSparkContext.parallelize(arrs);// 获取第一个元素Integer first = parallelize.first();// 打印结果System.out.println("第一个元素 = " + first);}
}

运行结果：

take操作

说明

在RDD中从头获取指定数量的元素，返回获取的元素集合。

案例

获取前3个元素

import lombok.extern.slf4j.Slf4j;
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 org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import scala.Tuple2;
import javax.annotation.PostConstruct;
import java.util.*;/*** spark 案例*/
@Slf4j
@Component
public class DemoTimer {@AutowiredJavaSparkContext javaSparkContext;@AutowiredSparkSession sparkSession;@PostConstructpublic void takeDemo(){// 创建一个集合List<Integer> arrs = Arrays.asList(1, 2, 3, 4, 5);// 创建RDDJavaRDD<Integer> parallelize = javaSparkContext.parallelize(arrs);// 获取前三个元素List<Integer> takes = parallelize.take(3);// 打印结果System.out.println("前三个元素 = " + takes);}
}

运行结果：

countByKey操作

说明

统计RDD中每种键的数量。

案例

根据键值对，统计键值对中不同键的数量。并打印。

import lombok.extern.slf4j.Slf4j;
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 org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import scala.Tuple2;
import javax.annotation.PostConstruct;
import java.util.*;/*** spark 案例*/
@Slf4j
@Component
public class DemoTimer {@AutowiredJavaSparkContext javaSparkContext;@AutowiredSparkSession sparkSession;@PostConstructpublic void countByKeyDemo(){// 创建一个集合List<Tuple2<String,Integer>> arrs = Arrays.asList(new Tuple2<String,Integer>("张三",15),new Tuple2<String,Integer>("张三",20),new Tuple2<String,Integer>("李四",20),new Tuple2<String,Integer>("李四",30),new Tuple2<String,Integer>("李四",50),new Tuple2<String,Integer>("王五",10));// 创建JavaPairRDD 对象，JavaPairRDD对象的元素为键值对。JavaPairRDD<String, Integer> parallelize = javaSparkContext.parallelizePairs(arrs);// 使用 countByKey() 方法，统计相同键的数量。Map<String, Long> countByKey = parallelize.countByKey();// 打印结果countByKey.forEach((key, value) -> System.out.println("键: " + key + ", 数量: " + value));}
}

运行结果：

saveAsTextFile 操作

说明

将 RDD 的数据以txt文件保存到外部存储系统。传入指定路径，文件会生成到该路径下。

注意
输出路径不能存在：如果输出路径已经存在，saveAsTextFile 会抛出异常
文件分区：Spark 会将每个分区的数据写入一个独立的文件。因此，如果 RDD 有多个分区，它会生成多个文件，每个文件对应一个分区的数据 。 文件名会根据分区编号进行自动命名，通常形式是：part-00***
文本格式：保存时，RDD 中的每个元素会被转换为文本行。默认情况下，Spark 会把 RDD 中的每个元素的 toString() 输出到文件中。
路径支持分布式存储：saveAsTextFile 支持将数据保存到本地文件系统、HDFS、S3 等分布式存储中。路径的格式取决于存储系统的类型。例如，如果要保存到 HDFS，路径应该以 hdfs:// 开头。

案例

将RDD存入外部文件。

以下​案例中，将基于集合生成的RDD保存到saveRddDemo文件夹中​。具体参考本文中的 RDD存入外部文件中 内容

import lombok.extern.slf4j.Slf4j;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import java.util.Arrays;
import java.util.List;/*** spark 案例*/
@Slf4j
@Component
public class DemoTimer {@AutowiredJavaSparkContext javaSparkContext;@AutowiredSparkSession sparkSession;/*** 创建 RDD 算子*/@PostConstructpublic void createRDD() {// 从集合创建 RDDList<String> lists = Arrays.asList("hello", "spark", "hi", "spark", "hadoop");// parallelize(): 创建RDD，RDD中创建默认分区数// parallelize( 元素, 分区数)： 创建RDD，RDD中指定分区数JavaRDD<String> parallelize = javaSparkContext.parallelize(lists); // 创建RDDparallelize.collect().forEach(System.out::println); // 打印// 存储的目录文件夹路径。此处为项目中的路径且目录必须为不存在的路径。String fileSavePath="I:\\zhang\\SpringBootSparkDemo\\src\\main\\resources\\saveRddDemo";parallelize.saveAsTextFile(fileSavePath);// 开始将RDD保存到文件中 }
}

foreach操作

说明

循环遍历RDD中的元素

案例

以上案例中大部分用到此方法，是否用方式看以上案例。