flink获取keyedstate值并在另一个流中使用

3phpmpom 于 2021-06-21 发布在 Flink

关注(0)|答案(3)|浏览(287)

我知道keyed状态属于its键，只有当前键访问它的状态值，其他键不能访问不同键的状态值。
我试图用相同的密钥访问状态，但在不同的流中。有可能吗？
如果这是不可能的话，我将有2个重复的数据？
not：我需要两个流，因为每个流都有不同的时间窗口和不同的实现。
下面是一个示例（我知道keyby（sommething）对于两个流操作都是相同的）：

public class Sample{
       streamA
                .keyBy(something)
                .timeWindow(Time.seconds(4))
                .process(new CustomMyProcessFunction())
                .name("CustomMyProcessFunction")
                .print();

       streamA
                .keyBy(something)
                .timeWindow(Time.seconds(1))
                .process(new CustomMyAnotherProcessFunction())
                .name("CustomMyProcessFunction")
                .print();
}

public class CustomMyProcessFunction extends ProcessWindowFunction<..>
{
    private Logger logger = LoggerFactory.getLogger(CustomMyProcessFunction.class);
    private transient ValueState<SimpleEntity> simpleEntityValueState;
    private SimpleEntity simpleEntity;

    @Override
    public void open(Configuration parameters) throws Exception
    {
        ValueStateDescriptor<SimpleEntity> simpleEntityValueStateDescriptor = new ValueStateDescriptor<SimpleEntity>(
                "sample",
                TypeInformation.of(SimpleEntity.class)
        );
        simpleEntityValueState = getRuntimeContext().getState(simpleEntityValueStateDescriptor);
    }

    @Override
    public void process(...) throws Exception
    {
        SimpleEntity value = simpleEntityValueState.value();
        if (value == null)
        {
            SimpleEntity newVal = new SimpleEntity("sample");
            logger.info("New Value put");
            simpleEntityValueState.update(newVal);
        }
        ...
    }
...
}

public class CustomMyAnotherProcessFunction extends ProcessWindowFunction<..>
{

    private transient ValueState<SimpleEntity> simpleEntityValueState;

    @Override
    public void open(Configuration parameters) throws Exception
    {

        ValueStateDescriptor<SimpleEntity> simpleEntityValueStateDescriptor = new ValueStateDescriptor<SimpleEntity>(
                "sample",
                TypeInformation.of(SimpleEntity.class)
        );
        simpleEntityValueState = getRuntimeContext().getState(simpleEntityValueStateDescriptor);
    }

    @Override
    public void process(...) throws Exception
    {
        SimpleEntity value = simpleEntityValueState.value();
        if (value != null)
            logger.info(value.toString()); // I expect that SimpleEntity("sample")
        out.collect(...);
    }
...
}

apache-flink flink-streaming

来源：https://stackoverflow.com/questions/62301565/flink-get-the-keyedstate-state-value-and-use-in-another-stream

3条答案

按热度按时间

flseospp1#

正如已经指出的，状态对于单个操作符示例总是局部的。不能共享。
但是，您可以做的是将状态更新从持有状态的操作符流式传输到其他需要它的操作符。通过侧输出，您可以创建复杂的数据流，而无需共享状态。

赞(0）回复(0）举报 2021-06-21

qncylg1j2#

为什么不能将状态作为Map操作的一部分返回，并且该流可以用于连接到其他流

赞(0）回复(0）举报 2021-06-21

c0vxltue3#

我试着用你的想法在两个操作员之间用同一个键共享状态。

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.io.IOException;

public class FlinkReuseState {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(3);

        DataStream<Integer> stream1 = env.addSource(new SourceFunction<Integer>() {
            @Override
            public void run(SourceContext<Integer> sourceContext) throws Exception {
                int i = 0;
                while (true) {
                    sourceContext.collect(1);
                    Thread.sleep(1000);
                }
            }

            @Override
            public void cancel() {

            }
        });

        DataStream<Integer> stream2 = env.addSource(new SourceFunction<Integer>() {
            @Override
            public void run(SourceContext<Integer> sourceContext) throws Exception {
                while (true) {
                    sourceContext.collect(1);
                    Thread.sleep(1000);
                }
            }

            @Override
            public void cancel() {

            }
        });

        DataStream<Integer> windowedStream1 = stream1.keyBy(Integer::intValue)
                .timeWindow(Time.seconds(3))
                .process(new ProcessWindowFunction<Integer, Integer, Integer, TimeWindow>() {
                    private ValueState<Integer> value;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        ValueStateDescriptor<Integer> desc = new ValueStateDescriptor<Integer>("value", Integer.class);
                        value = getRuntimeContext().getState(desc);
                    }

                    @Override
                    public void process(Integer integer, Context context, Iterable<Integer> iterable, Collector<Integer> collector) throws Exception {
                        iterable.forEach(x -> {
                            try {
                                if (value.value() == null) {
                                    value.update(1);
                                } else {
                                    value.update(value.value() + 1);
                                }
                            } catch (IOException e) {
                                e.printStackTrace();
                            }
                        });
                        collector.collect(value.value());
                    }
                });

        DataStream<String> windowedStream2 = stream2.keyBy(Integer::intValue)
                .timeWindow(Time.seconds(3))
                .process(new ProcessWindowFunction<Integer, String, Integer, TimeWindow>() {

                    private ValueState<Integer> value;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        ValueStateDescriptor<Integer> desc = new ValueStateDescriptor<Integer>("value", Integer.class);
                        value = getRuntimeContext().getState(desc);
                    }

                    @Override
                    public void process(Integer s, Context context, Iterable<Integer> iterable, Collector<String> collector) throws Exception {
                        iterable.forEach(x -> {
                            try {
                                if (value.value() == null) {
                                    value.update(1);
                                } else {
                                    value.update(value.value() + 1);
                                }
                            } catch (IOException e) {
                                e.printStackTrace();
                            }
                        });
                        collector.collect(String.valueOf(value.value()));
                    }
                });

        windowedStream2.print();

        windowedStream1.print();

        env.execute();

    }
}

它不工作，每个流只更新自己的值状态，输出如下所示。

键控状态
根据官方文档，*每个键控状态逻辑上绑定到 <parallel-operator-instance, key> ，因为每个键“属于”一个键控操作符的一个并行示例，我们可以简单地认为这是 <operator, key>* .
我认为不可能通过给不同运营商的州起相同的名字来共享州。
你试过协处理函数吗？这样，您还可以为每个流实现两个进程函数，唯一的问题就是时间窗口。你能提供更多关于你的过程逻辑的细节吗？

赞(0）回复(0）举报 2021-06-21

我来回答

flink获取keyedstate值并在另一个流中使用

3条答案

相关问题

热门标签

最新问答