saprk 算子reduce和reduceByKey的区别

`reduceByKey`是Spark RDD的一个转换操作，用于对具有相同键的元素进行聚合。它将RDD中的每个键值对聚合成一个单独的值，并返回一个新的RDD。该操作需要一个作为参数的函数，该函数将两个值合并为一个值。 例如，如果我们有一个包含键值对的RDD，其中每个键都有多个值： ``` (1, 2) (1, 4) (2, 1) (2, 3) (2, 5) ``` 我们可以使用`reduceByKey`对每个键进行聚合，例如求和操作： ```python rdd = sc.parallelize([(1, 2), (1, 4), (2, 1), (2, 3), (2, 5)]) result = rdd.reduceByKey(lambda x, y: x + y) ``` 执行上述操作后，我们将得到一个新的RDD，其中每个键都对应于一个值，该值是该键对应的所有值的总和： ``` (1, 6) (2, 9) ``` 在上面的例子中，`reduceByKey`操作将键1的值2和4相加，将键2的值1、3和5相加，并分别将结果6和9与相应的键1和2一起返回。

在 IDEA 中实现 Spark 相关任务（包括 RDD 基本算子实验和 WordCount 综合实验），需要先设置好开发环境，再基于具体的业务需求编写代码。下面详细介绍如何实现实验内容： ### 第一部分：掌握 RDD 算子的使用 #### 实现步骤 1. **准备环境** - 创建一个 Maven 工程。 - 修改 `pom.xml` 添加 Spark 依赖： ```xml <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-core_2.12</artifactId> <version>3.0.1</version> </dependency> ``` 2. **初始化 SparkSession 和 SparkContext** 这是所有 Spark 操作的基础。 ```java import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; public class RddOperatorDemo { private static final String SPARK_MASTER_URL = "local"; public static void main(String[] args) { // 初始化 Spark Context SparkConf conf = new SparkConf().setAppName("RddOperator").setMaster(SPARK_MASTER_URL); JavaSparkContext sc = new JavaSparkContext(conf); rddOperators(sc); // 调用算子练习函数 sc.close(); // 关闭上下文 } /** * 测试各种 RDD 转换算子与行动算子的功能 */ private static void rddOperators(JavaSparkContext sc) { List<String> wordsList = Arrays.asList("hello", "world", "hello spark", "hadoop"); // 生成初始 RDD JavaRDD<String> initialRDD = sc.parallelize(wordsList); // map 示例：将每个单词转成大写形式 System.out.println("--- Map Operator ---"); JavaRDD<String> mappedRDD = initialRDD.map(s -> s.toUpperCase()); mappedRDD.collect().forEach(System.out::println); // distinct 示例：去重操作 System.out.println("\n--- Distinct Operator ---"); JavaRDD<String> distinctRDD = initialRDD.distinct(); distinctRDD.collect().forEach(System.out::println); // flatMap 示例：对字符串按空格拆分出单个词语集合 System.out.println("\n--- FlatMap Operator ---"); JavaRDD<String> flattenWordsRDD = initialRDD.flatMap(s -> Arrays.asList(s.split(" ")).iterator()); flattenWordsRDD.collect().forEach(System.out::println); // filter 示例：过滤掉长度小于等于4的元素 System.out.println("\n--- Filter Operator ---"); JavaRDD<String> filteredRDD = flattenWordsRDD.filter(word -> word.length() > 4); filteredRDD.collect().forEach(System.out::println); // reduceByKey & mapValues 示例：词频统计前处理阶段 System.out.println("\n--- ReduceByKey and MapValues Operators ---"); JavaPairRDD<String, Integer> pairRDD = flattenWordsRDD.mapToPair(word -> new Tuple2<>(word, 1)); JavaPairRDD<String, Integer> reducedRDD = pairRDD.reduceByKey((v1, v2) -> v1 + v2); reducedRDD.collect().forEach(tuple -> System.out.println(tuple._1 + ": " + tuple._2)); // groupByKey 示例：按照 key 分组值列表 System.out.println("\n--- GroupByKey Operator ---"); JavaPairRDD<String, Iterable<Integer>> groupedRDD = pairRDD.groupByKey(); groupedRDD.collect().forEach(entry -> System.out.printf("%s => %s\n", entry._1(), Lists.newArrayList(entry._2()))); // sortByKey 示例：按键排序键值对 System.out.println("\n--- SortByKey Operator ---"); JavaPairRDD<Integer, String> invertedPairsRDD = flattenedWordsRDD.mapToPair(word -> new Tuple2<>(word.hashCode(), word)); invertedPairsRDD.sortByKey(true).collect().forEach(pair -> System.out.println(pair._2())); } } ``` --- ### 第二部分：Spark 算子综合实验 —— WordCount （词频统计） #### 实现步骤 1. 同样在上面基础上继续新增一段针对 WordCount 的完整代码逻辑。 ```java public static void runWordCountExample(JavaSparkContext sc){ String inputFilePath = "/path/to/input/text/file"; // 替换为实际输入路径 JavaRDD<String> linesRDD = sc.textFile(inputFilePath); // 加载文本数据作为行级 RDD // 单词分割、映射到(key,value) JavaRDD<String> allWords = linesRDD.flatMap(line -> Arrays.asList(line.toLowerCase().split("\\W+")).iterator()); JavaPairRDD<String,Integer> pairs = allWords.mapToPair(word->new Tuple2<>(word,1)); // 计算每种词汇的数量总计 JavaPairRDD<String,Integer> counts = pairs.reduceByKey(Integer::sum); // 输出最终结果集 counts.foreach(tup->{System.out.println(tup._1()+" : "+tup._2());}); } ``` 2. 最后记得修改主方法调用新加入的任务入口： ``` public static void main(String[] args) { ... runWordCountExample(sc); ...} ```