RichFlatMap中的状态管理有和没有键通过

Question

我有这样一个流应用程序：

DataStream<MyObject> stream1 = source
                .keyBy("clientip")
                .flatMap(new MyFlatMapFunction())
                .name("Stream1");

//...
public class MyFlatMapFunction extends RichFlatMapFunction<MyObject, MyObject> {

    private transient ValueState<Boolean> valueState;

    @Override
    public void open(Configuration parameters)
    {
        StateTtlConfig ttlConfig = StateTtlConfig
                .newBuilder(Time.minutes(12))
                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                .setStateVisibility(StateTtlConfig.StateVisibility.ReturnExpiredIfNotCleanedUp).cleanupInBackground()
                .build();

        ValueStateDescriptor<Boolean> valueStateeDescriptor = new ValueStateDescriptor<>(
                "valueState",
                Types.BOOLEAN);

        valueStateeDescriptor.enableTimeToLive(ttlConfig);
        valueState = getRuntimeContext().getState(valueState);
    }

    @Override
    public void flatMap(MyObject myObject, Collector<MyObject> collector) throws Exception
    {
       // get value from value state, check if it is matched with something
       // if matches some condition, then collector.collect(myObject)
       // update state for each myObject
    }
}

否：3台机器上有3名工人，并行度为16。总并行度为48。

在实现此代码时，我始终假设“如果ip地址1.2.3.4与条件匹配，则来自同一ip地址1.2.3.4的后续请求始终与条件匹配，直到状态被清除”。这句话对吗？

据我从flink docs了解，如果ip地址1.2.3.4转到机器1（通过生成clientip的哈希值），那么来自ip地址1.2.3.4的所有请求都会转到机器1？

方法在taskmanager jvm中调用一次。因此，flink创建了48个flatMapOperation实例（48个中的1-15个驻留在machine1中，48个中的16-32个驻留在machine2中，48个中的33-48个驻留在machine3中），每个flatMapInstance都将运行open方法。这意味着开放式方法需要48次？

最后，所有48个实例都访问相同的状态，但值不同（因为状态是本地的）。我的意思是，一部分实例组（假设机器1上的16个实例）将获得相同的状态值。

最后，如果在FlatMap之前没有keyBy，那么来自ip地址的请求1.2.3.4可以随机访问机器1、机器2或机器3？

Answer 1

1. 由于您执行了一个keyBy（“clientip”），因此该字段具有相同值的所有记录将由相同的MyFlatMapFunction子任务处理。因此，所有记录的集合被划分为48个子任务，并且假设IP地址的计数均匀分布，每个子任务将获得大约48条记录的1/48
2. 是的，将有48个MyFlatMapFunction的实例被实例化，因此有48个对open（）的调用
3. 所有48个实例都访问相同的状态。否，状态是每个唯一键，因此状态是按键值在48个子任务之间划分的
4. 如果没有keyBy（），那么MyFlatMapFunction操作符的每个子任务都将从源中获取分区中的任何数据。这取决于您的数据源，例如，如果您正在阅读一个Kafka主题，而该主题有48个分区，那么从Kafka分区到MyFlatMapFunction子任务有一个1对1的映射。如果您的Kafka分区少于48个，那么您的一些MyFlatMapFunction子任务将无法获取任何数据。如果要将传入记录重新分发到所有子任务，则可以执行重新平衡（）。但请注意，您将无法维护每个IP地址的状态