Flink 将所有流元素保存在 HashMap 中答案

【问题标题】：Flink keeping all stream elements in HashMapFlink 将所有流元素保存在 HashMap 中
【发布时间】：2019-11-26 22:05:05
【问题描述】：

我正在尝试在 Flink 作业的窗口函数中使用 HashMap。是否可以将所有并行算子的所有元素都存储在一个算子上的 HashMap 中？

public class SeewoUserWindowFunction implements WindowFunction<ObjectNode, LabelInfo, String, TimeWindow> {

    private static final Logger logger = LoggerFactory.getLogger(SeewoUserWindowFunction.class);
    @Override
    public void apply(String s, TimeWindow timeWindow, Iterable<ObjectNode> iterable, Collector<LabelInfo> collector) throws Exception {
        try {
            HashMap<String, LabelInfo> result = new HashMap<>();
            iterable.forEach(e -> {
                    String key = e.get("value").get("$tid").toString() + "/" + e.get("value").get("$code").toString();
                    if (result.containsKey(key)) {
                        result.put(key, result.get(key).update(e, timeWindow.getEnd()));
                    } else {
                        result.put(key, LabelInfo.of(e, timeWindow.getEnd()));
                    }
            });
            result.values().stream().forEach(labelInfo -> collector.collect(labelInfo));
        } catch (Exception exception) {
            logger.error("parse exception!", exception);
        }
    }
}

【问题讨论】：

嘿，我想我不明白你的问题。我在您的代码中没有看到任何单例。
嘿，对不起我的英语。我的意思是，如果我为 n (n > 1) 设置 flink 运算符的并行性，并且我为计数数据定义一个哈希图。这个 hashmap 可以记录所有数据，而不是只记录 n 个并行算子数据中的一个。换句话说，我的问题是并行算子如何交换数据？谢谢

标签： java parallel-processing apache-flink

【解决方案1】：

在您的情况下，每个并行运算符都会简单地保留它自己的HashMap，但这在很大程度上取决于您的流的分区。有类似的问题可以解释运营商here 之间的通信。如果您出于某种原因希望将流的所有元素保留在HashMap 中并使用parallelism > 1。您可以在您的数据流上调用global()，这将导致流的所有元素仅通过并行运算符的一个实例，这基本上允许您将所有流元素存储在HashMap中，但请记住这可能会在吞吐量和延迟方面产生可怕的后果。

【讨论】：

【解决方案2】：

您可以使用org.apache.flink.streaming.api.datastream.DataStream#windowAll 方法将所有元素聚集到一个全局窗口中。
见this document。

【讨论】：