Flink:在CoFlatMapFunction中共享状态

Question

使用CoFlatMapFunction时有点卡住了。如果我把它放在DataStreambefore窗口上，它似乎可以正常工作，但如果放在窗口的“apply”函数之后，它就会失败。

我正在测试两个流，主功能在flatMap1不断摄取数据和控制流模型在flatMap2更改模型的要求。

我能够在flatMap2中正确设置并查看b0/b1，但是flatMap1始终可以看到b0和b1在初始化时设置为0。

我是不是漏了什么明显的东西？

public static class applyModel implements CoFlatMapFunction<Features, Model, EnrichedFeatures> {
    private static final long serialVersionUID = 1L;

    Double b0;
    Double b1;

    public applyModel(){
        b0=0.0;
        b1=0.0;
    }

    @Override
    public void flatMap1(Features value, Collector<EnrichedFeatures> out) {
        System.out.print("Main: " + this + "\n");
    }

    @Override
    public void flatMap2(Model value, Collector<EnrichedFeatures> out) {
        System.out.print("Old Model: " + this + "\n");
        b0 = value.getB0();
        b1 = value.getB1();
        System.out.print("New Model: " + this + "\n");
    }

    @Override
    public String toString(){
        return "CoFlatMapFunction: {b0: " + b0 + ", b1: " + b1 + "}";
    }
}

Answer 1

以下是邮件列表中的答案。。。

CoFlatMapFunction是否打算并行执行？

如果是，则需要某种方法来确定地将哪个记录分配给哪个并行实例。在某种程度上，CoFlatMapFunction在模型和会话窗口的结果之间进行并行（分区）连接，因此您需要某种形式的键来选择元素转到哪个分区。这有意义吗？

如果不是，请尝试显式地将其设置为parallelism 1。

你好斯蒂芬

所有人都可以只读访问的全局状态可以通过广播（）实现。

可供所有人读取和更新的全局状态目前不可用。一致的操作将是相当昂贵的，需要某种形式的分布式通信/共识。

相反，我鼓励您采用以下方法：

1）如果可以对状态进行分区，请使用keyBy（）。mapBackState（）-这可以本地化状态操作，并使其非常快。

2）如果你的状态不是按键组织的，你的状态很可能很小，你可能可以使用非并行操作。

3）如果一些操作更新了状态，另一个操作访问了状态，您通常可以通过迭代和CoFlatMapFunction来实现（一边是原始输入，另一边是反馈输入）。

所有的方法最终都将状态访问和修改本地化，如果可能的话，这是一个很好的模式。

你好斯蒂芬