Apache Flink：如何计算 DataStream 中的事件总数

Posted 2023-04-18

【中文标题】Apache Flink：如何计算 DataStream 中的事件总数

【英文标题】：Apache Flink: How to count the total number of events in a DataStream

【发布时间】：2017-11-12 08:04:24

【问题描述】：

我有两个原始流，我正在加入这些流，然后我想计算已加入的事件总数和未加入的事件总数。我通过在joinedEventDataStream 上使用地图来做到这一点，如下所示

joinedEventDataStream.map(new RichMapFunction<JoinedEvent, Object>() 

            @Override
            public Object map(JoinedEvent joinedEvent) throws Exception 

                number_of_joined_events += 1;

                return null;
            
        );

问题 #1：这是计算流中事件数量的合适方法吗？

问题#2：我注意到了一种有线行为，你们中的一些人可能不相信。问题是，当我在 IntelliJ IDE 中运行我的 Flink 程序时，它显示了 number_of_joined_events 的正确值，但在我将此程序提交为 jar 的情况下，它显示了 0 的正确值。因此，当我将程序作为jar 文件而不是实际计数运行时，我得到了number_of_joined_events 的初始值。为什么这种情况只发生在jar文件提交的情况下，而不是在IDE中？

【参考方案1】：

您的方法不起作用。您在通过 JAR 文件执行程序时注意到的行为是预期的。

我不知道number_of_joined_events 是如何定义的，但我假设它在您的程序中是一个静态变量。当您在 IDE 中运行程序时，它会在单个 JVM 中运行。因此，所有操作员都可以访问静态变量。当您将 JAR 文件提交到远程进程时，程序会在不同的 JVM（可能是多个 JVM）中执行，并且您的客户端进程中的静态变量永远不会更新。

您可以使用 Flink 的指标或将 1s 相加的 ReduceFunction 来计算已处理记录的数量。

【讨论】：

哇，你的天才。它是一个静态变量，我在joinedDataStrem 中使用了一个累加器，它在jar 文件中工作。不过我花了我的周末:)。非常感谢