java - 了解 JavaPairRDD.reduceByKey 函数

Question

我遇到了 Apache Spark 的以下代码片段:

JavaRDD<String> lines = new JavaSparkContext(sparkSession.sparkContext()).textFile("src\\main\\resources\\data.txt");
JavaPairRDD<String, Integer> pairs = lines.mapToPair(s -> new Tuple2(s, 1));
System.out.println(pairs.collect());
JavaPairRDD<String, Integer> counts = pairs.reduceByKey((a, b) -> a + b);
System.out.println("Reduced data: " + counts.collect());

我的data.txt如下:

Mahesh
Mahesh
Ganesh
Ashok
Abnave
Ganesh
Mahesh

输出为:

[(Mahesh,1), (Mahesh,1), (Ganesh,1), (Ashok,1), (Abnave,1), (Ganesh,1), (Mahesh,1)]
Reduced data: [(Ganesh,2), (Abnave,1), (Mahesh,3), (Ashok,1)]

虽然我理解如何获得第一行输出，但我不明白如何获得第二行，即JavaPairRDD<String, Integer> counts由 reduceByKey 组成.

我发现signature的reduceByKey()如下:

public JavaPairRDD<K,V> reduceByKey(Function2<V,V,V> func)

Function2.call() 的[签名]( http://spark.apache.org/docs/1.2.0/api/java/org/apache/spark/api/java/function/Function2.html#call(T1 , T2))如下:

R call(T1 v1, T2 v2) throws Exception

reduceByKey()的解释内容如下:

Merge the values for each key using an associative reduce function. This will also perform the merging locally on each mapper before sending results to a reducer, similarly to a "combiner" in MapReduce. Output will be hash-partitioned with the existing partitioner/ parallelism level.

现在这个解释对我来说听起来有点令人困惑。 reduceByKey() 的功能可能还有更多功能。通过查看 reduceByKey() 的输入和输出和Function2.call() ，我感觉不知何故reducebyKey()将相同键的值发送到 call()成对。但这听起来并不明确。谁能解释一下到底如何reduceByKey()和Function2.call()一起工作吗？

Answer 1

顾名思义，reduceByKey() 根据您传递给它的 lambda 函数来减少数据。

在您的示例中，此函数是一个简单的加法器:对于 a 和 b，返回 a + b。了解结果如何形成的最好方法是想象内部发生的情况。 ByKey() 部分根据键值对记录进行分组。在您的示例中，您将有 4 组不同的对:

第 1 组:((Mahesh, 1), (Mahesh, 1), (Mahesh, 1))

第二组:((Ganesh, 1), (Ganesh, 1))

第 3 组:((Ashok, 1))

第 4 组:((Abnave, 1))

现在，reduce 部分将尝试使用 lambda 函数(加法器)来减少之前的 4 个集合:

对于第 1 组:(Mahesh, 1 + 1 + 1) -> (Mahesh, 3)

对于第 2 组:(Ganesh, 1 + 1) -> (Ganesh, 2)

对于第 3 组:(Ashok，1) -> (Ashok，1)(无需添加)

对于第 4 组:(Abnave, 1) -> (Abnave, 1)(无需添加)

函数签名有时可能会令人困惑，因为它们往往更通用。