如何针对特定用例在 reduceByKeyAndWindow() 中实现 invFunc

Question

我正在使用火花流来处理文件流。多个文件成批到达并从所有文件中激发过程数据。 我的用途是获取进入后续批次的文件的每条记录的总和。例如：

• key: key_1 value: 10 --> batch1
• key: key_1 value: 05 --> batch1
• key: key_1 value: 19 --> batch2
• key: key_1 value: 11 --> batch3
• key: key_1 value: 10 --> batch4

我需要如下输出：

• 处理第一批后，我需要输出为 => key: key_1 val: 15
• 处理第二批后，我需要输出为 => key: key_1 val: 34
• 处理第三批后，我需要输出为 => key: key_1 val: 45
• 处理第 4 批后，我需要输出为 => key: key_1 val: 55
• 处理第 5 批后，我需要输出为 => key: key_1 val: 55

我的 reduceByKeyAndWindow() 代码如下：

JavaPairDStream<String, Summary> grpSumRDD = sumRDD.reduceByKeyAndWindow(GET_GRP_SUM, Durations.minutes(2*batchInterval), Durations.minutes(batchInterval));

private static final Function2<Summary, Summary, Summary> GET_GRP_SUM = new Function2<Summary, Summary, Summary>() {
    private static final long serialVersionUID = 1L;

    public Summary call(Summary s1, Summary s2) throws Exception {
        try {

            Summary s = new Summary();

            long grpCnt = s1.getDelta() + s2.getDelta();
            s.setDeltaSum(grpCnt);

            return s;
        } catch (Exception e) {
            logger.error(" ==== error in CKT_GRP_SUM ==== :"+e);
            return new Summary();
        }
    }

};

我从上面的实现中得到的输出如下：

• 处理第一批后，我得到输出 => key: key_1 value: 15
• 处理第二批后，我得到输出 => key: key_1 value: 34
• 处理第三批后，我得到输出 => key: key_1 value: 30
• 处理第 4 批后，我得到输出 => key: key_1 value: 21
• 处理第 5 批后，我得到输出 => key: key_1 value: 10

根据 reduceByKeyAndWindow() 的输出，它似乎正在计算先前批次数据和当前批次数据的聚合。 但我的要求是对上一批的聚合数据和当前的批数据进行聚合。这样根据上面的例子 它应该在第 4 和第 5 批结束时输出为 [(((15)+19)+11)+10 = 55]。

我读到了 reduceByKeyAndWindow() 和 invFunc 可以实现以获得预期的输出。我试图实现它类似于 GET_GRP_SUM 但它没有给我预期的结果。任何有关正确实施以获得所需输出的帮助将不胜感激。

我正在使用 java 1.8.45 和 spark 版本 1.4.1 和 hadoop 版本 2.7.1。

我用 reduceByKeyAndWindow() 对 invFunc 的实现

JavaPairDStream<String, Summary> grpSumRDD = sumRDD.reduceByKeyAndWindow(GET_GRP_SUM, INV_GET_GRP_SUM, Durations.minutes(2*batchInterval), Durations.minutes(batchInterval));

private static final Function2<Summary, Summary, Summary> INV_GET_GRP_SUM = new Function2<Summary, Summary, Summary>() {
    private static final long serialVersionUID = 1L;

    public Summary call(Summary s1, Summary s2) throws Exception {
        try {

            Summary s = new Summary();

            long grpCnt = s1.getDelta() + s2.getDelta();
            s.setDeltaSum(grpCnt);

            return s;

        } catch (Exception e) {
            logger.error(" ==== error in INV_GET_GRP_SUM ==== :"+e);
            return new Summary();
        }
    }
};

我已经像上面那样实现了我的 invFunc，这并没有给我预期的输出。我这里分析的是 s1 和 s2 给我之前批次的聚合值，我觉得我不太确定。

我尝试更改我的 invFunc 实现，如下所示：

private static final Function2<Summary, Summary, Summary> INV_GET_GRP_SUM = new Function2<Summary, Summary, Summary>() {
    private static final long serialVersionUID = 1L;

    public Summary call(Summary s1, Summary s2) throws Exception {
        try {

            return s1;

        } catch (Exception e) {
            logger.error(" ==== error in INV_GET_GRP_SUM ==== :"+e);
            return new Summary();
        }
    }
};

这个实现给了我预期的输出。但我面临的问题是带有 invFunc 的 reduceByKeyAndWindow() 不会自动删除旧键。我又看了几篇文章，发现我需要编写自己的过滤器函数，该函数将删除具有 0 值（无值）的旧键。

再次，我不确定如何编写过滤函数来删除具有 0 值（无值）的旧键，因为我没有具体了解 s1 和 s2 返回到 INV_GET_GRP_SUM 的内容。

Answer 1

使用 UpdateStateByKey

您是否已从 Streaming API 签出 updateStateByKey()？它允许您在批处理间隔之间维护键值对的状态，并使用与之关联的新信息（值）不断更新每个键。这适用于您的用例，因为之前的数据状态将包含每个键的聚合总和，直到最新状态。有关此功能的更多信息，请参阅其用法here 和示例here。

关于该函数的一个注意事项是，它需要启用检查点，以便可以在每次迭代时保存状态。

（编辑：）

使用 ReduceByKeyAndWindow

关于使用reduceKeyAndWindow()，您的普通func 和invFunc 的call() 方法的第二个参数是添加的新元素和旧元素分别减去的元素。本质上，您是通过在新时间片上添加元素（您正在使用GET_GRP_SUM）并从旧时间片中减去元素（您没有使用INV_GET_GRP_SUM）来实现此窗口化缩减。请注意，在您的第一次尝试中，您将旧值重新添加回当前窗口中的值，而在第二次尝试中，您将忽略移出窗口的值。

要从移出窗口的元素中减去旧值，您可能希望 INV_GET_GRP_SUM 具有类似于以下的逻辑（并且可以找到类似的正确实现 here）：

public Summary call(Summary s1, Summary s2) throws Exception {
    try {

        long grpCnt = s1.getDelta() - s2.getDelta();
        s.setDeltaSum(grpCnt);

    } catch (Exception e) {
        logger.error(" ==== error in INV_GET_GRP_SUM ==== :"+e);
        return new Summary();
    }
}

对于您的另一个问题，似乎确实有一种方法可以过滤掉过期的密钥，并且正如您所提到的，它确实涉及编写过滤器函数。正如您从API 中看到的那样，此过滤器函数接受您的键值对并返回一个布尔值，该值将设置为 true（如果您想保留该对）或 false（如果您想删除该对） .在这种情况下，由于您想在值为零时删除您的配对，您可以执行以下操作：

private static final Function<scala.Tuple2<String, Summary>, Boolean> FILTER_EXPIRED = new Function<scala.Tuple2<String, Summary>, Boolean>() {
    public Boolean call(scala.Tuple2<String, Summary> s) { 
        return s.productElement(1) > 0; 
    }
}

然后你可以把它传递给你的reduceByKeyAndWindow()函数（注意你应该在这里传入分区参数来确定你的DStream中的RDD将使用多少个分区）：

JavaPairDStream<String, Summary> grpSumRDD = sumRDD.reduceByKeyAndWindow(GET_GRP_SUM, INV_GET_GRP_SUM, Durations.minutes(2*batchInterval), Durations.minutes(batchInterval), partitions, FILTER_EXPIRED);