java - Apache Spark 中按键不同的 Reducer

Question

我有一个大规模数据处理问题，我正尝试使用 Java 中的 Apache Spark 来解决。我的输入是一大组相对较小的自定义 Java 对象。

我的 map 步骤对每个对象进行了一些小改动。完成后，它会识别该对象所属的一个或多个等价类。组合起来，可能有数十亿个等价类/对象对。

我的问题是我需要对每个等价类的元素进行不同的操作。这个项目需要支持插件架构，所以我不知道等价类是什么，或者每个类需要发生的不同操作。

我的直觉是使用如下内容:

//Get the input set.
JavaRDD<MyType> input = ... //Not important

//Transform the input into (Equivalence Class, MyType) pairs, 
//using strings to store the equivalence class.
JavaPairRDD<String, MyType> classedInput = input.flatMapToPair(

    new PairFlatMapFunction<MyType, String, MyType>() {

        Iterator<Tuple2<String, MyType>> call(MyType arg) {

            List<Tuple2<String, MyType>> out = new ArrayList<>();

            //Compute equivalence classes for arg.
            for(String eqClz: getEquivalenceClasses(arg)) {
                out.add(new Tuple2<String, MyType>(equClz, arg));
            }

            return out.iterator();
        }
});

//Collapse the results for each equivalence class.
JavaPairRDD<String, MyType> output = classedInput.reduceByKey(

    new Function2<MyType, MyType, MyType>() {

        MyType call(MyType a, MyType b) {
            String eqClz = ??? //<= Problem
            List<MyModule> modules = MyFramework.getModulesForEqClz(eqClz);
            for(MyModule m: modules) {
               a = m.merge(a, b);
            }
            return a;
        }
    }

);

我希望能够将等价类传递给 reduceByKey 函数，以便使用它来确定需要调用哪些模块。问题是我找不到的 Spark 键控组合器函数都没有将 key 传递给它们的回调。

由于 classedInput 的大小，我想避免使用 MyType 对象保存 key ，或者在映射后添加太多额外的分布式操作。

有没有更像 Spark 的方法来完成我正在尝试的事情？

Answer 1

看来你的问题与"secondary sort"相反问题。我认为它可以通过逆解来解决(下面 2.)。

1. 一种方法是使用 reduce函数(或其更完整的版本，aggregate)，它只需要一个关联操作来汇总数据的结果，而不管键是什么。但是，在同一等价类中表达项目分组的详细信息可能会有点复杂。
2. 保持对项目匹配的等价类的引用的更简单方法是简单地在值中重复等价类:


Tuple2<String, MyType> outValue = new Tuple2<String, MyType>(eqClz, arg);
out.add(new Tuple2<String, Tuple2<String, MyType>>(equClz, outValue));


如果正如您在评论中提到的那样，您担心传输数据时随机播放的大小，那么您可能想要限制的是用作键控结构的表示的大小。我的意思是，按照上面的建议重复值中的键会导致 equClz 变量的两个副本。但是如果是十几个字节，你要缩小的就是关键位置的副本。为此，您可能需要选择长度合适的非加密散列。

您提到每条记录多十几个字节会导致多千兆字节的数据，这应该意味着您最多有几亿条记录，因此最多有几亿个“equClz”值。这很容易被 32 位非加密散列覆盖(您会很容易找到这些的实现，Murmur3，XXHash)。由于 32 位是 4 个字节，它应该将您的传输开销减少至少一个数量级。