Apache Spark - reducebyKey - Java -

Question

我正在尝试使用 java 作为编程语言来了解 Spark 中 reduceByKey 的工作原理。

假设我有一个句子“我就是我”。 我将句子分解成单词并将其存储为列表[I, am, who, I, am]。

现在这个函数将1分配给每个单词：

JavaPairRDD<String, Integer> ones = words.mapToPair(new PairFunction<String, String, Integer>() {
    @Override
    public Tuple2<String, Integer> call(String s) {
        return new Tuple2<String, Integer>(s, 1);
    }
});

所以输出是这样的：

(I,1) 
(am,1)
(who,1)
(I,1)
(am,1)

现在，如果我有 3 个 reducer 正在运行，每个 reducer 都会获得一个键以及与该键关联的值：

reducer 1:
    (I,1)
    (I,1)

reducer 2:
    (am,1)
    (am,1)

reducer 3:
    (who,1)

我想知道

一个。下面的函数中究竟发生了什么。
湾。参数是什么new Function2<Integer, Integer, Integer>
C。基本上JavaPairRDD是如何形成的。

JavaPairRDD<String, Integer> counts = ones.reduceByKey(new Function2<Integer, Integer, Integer>() {
    @Override
    public Integer call(Integer i1, Integer i2) {
        return i1 + i2;
    }
});

Answer 1

我认为您的问题围绕这里的 reduce 函数，这是一个返回 1 的 2 个参数的函数，而在 Reducer 中，您实现了一个多对多的函数。

如果不太通用，这个 API 会更简单。在这里，您提供了一个关联操作，可以将任何 2 个值减少到 1（例如，两个整数之和为 1）。这用于将每个键的所有值减少到 1。没有必要提供 N 对 1 功能，因为它可以通过 2 对 1 功能来完成。在这里，您不能为一个键发出多个值。

结果是每个（键，一组值）的（键，减少的值）。

经典 Hadoop MapReduce 中的 Mapper 和 Reducer 实际上都非常相似（只是采用一组值而不是每个键的单个值）并允许您实现许多模式。以一种很好的方式，以一种既浪费又复杂的方式。

您仍然可以重现 Mappers 和 Reducers 所做的事情，但 Spark 中的方法是 mapPartitions，可能与 groupByKey 配对。这些是您可能会考虑的最通用的操作，我并不是说您应该在 Spark 中以这种方式模拟 MapReduce。事实上，它不太可能是有效的。但这是可能的。

Answer 2

reduceByKey 的工作原理如下：

在 RDD 中，如果 spark 找到具有相同键的元素，则 spark 获取它们的值并对这些值执行某些操作，并返回相同类型的值。例如，让我们看看，你有和 RDD 元素：

[k,V1],[K,V2],这里V1,V2是f同类型 那么 new Function2() 的参数可以是三个。

1. 来自第一个 K,V 对的值部分，即 V1。
2. 来自第二个 K,V 对的值部分，即 V2。
3. 被覆盖的调用方法的返回类型，它又是 类型 V1 和 V2（可以是提供的函数操作的结果 调用方法的一部分）。

注意，由于RDD分布在节点间，每个节点都会执行自己的reduce操作，并将结果返回给master，master再对worker的结果进行最终的reduce操作。

我想这解释了你的问题。

Answer 3

reduceByKey，顾名思义，会对键相同的 JavaPairRDD 应用 reduce 操作。如果您参考文档，它会说 reduceByKey

使用关联和交换归约函数合并每个键的值。

reduceByKey 需要一个 Function2 接口的实现。 Function2 的语法是：Function2<T1, T2, R> 这里，输入参数的类型为 T1 和 T2，输出参数的类型为 R。

让我们通过您提到的示例来理解这一点

您要应用 reduceByKey 的 JavaPairRDD 是：

(I,1) 
(am,1)
(who,1)
(I,1)
(am,1)

在您的 JavaPairRDD 中，Key 是第一个参数（在本例中是单词），值是第二个参数（分配给每个单词的 1）。您想应用 reduceByKey 以了解哪个单词出现了多少次。每当我们看到相同的单词时，我们都想将 JavaPairRDD 的值相加。因此，要将值相加，您需要两个输入参数，返回值将是一个参数。

因此前两个整数，语法中的整数是指输入，第三个整数是指输出。与 Function2 接口的语法相关，T1 和 T2 是整数，R 也是整数。

回答问题 c)

应用reduceByKey操作最终形成的JavaPairRDD将以key作为应用reduceByKey的JavaPairRDD的原始key，value将是在 Function2 接口的实现中计算的最终缩减值。

如果您对这些功能接口的参数感到困惑。那么你大概可以使用这个规则：输入参数后跟输出参数在接口的语法声明中。

输入参数/参数将在函数的括号中，输出参数将是函数名称之前提到的参数。

例如：

1. 查看您提出的问题中的 PairFunction 声明。是PairFunction<String, String, Integer>，对应的调用方式是 Tuple2<String, Integer> call(String s) 。所以这里输入为String，输出由String、Integer组成。
2. 查看 Function2 接口的声明。为Function2<Integer, Integer, Integer>，对应的调用方式为Integer call(Integer i1, Integer i2)。所以输入是两个整数，输出是一个整数。

我希望它有所帮助。

Answer 4

简而言之，考虑一下：

输入：{(a:1),(b:2),(c:2),(a:3),(b:2),(c:3)}

将其传递给reduceByKey。

输出：{(a:4),(b:4),(c:5)}