Flink flatMap() - NullPointerException

Question

以下是代码：

public class VerifyDuplicate {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        SingleOutputStreamOperator<Tuple3<String, String, Integer>> dataStream = env.fromElements(
            Tuple3.of("ID_1", "subid_1", 1),
            Tuple3.of("ID_2", "subid_2", 2),
            Tuple3.of("ID_3", "subid_3", 3),
            Tuple3.of("ID_4", "subid_4", 4),
            Tuple3.of("ID_4", "subid_4", 4),
            Tuple3.of("ID_6", "subid_6", 6),
            Tuple3.of("ID_4", "subid_7", 7),
            Tuple3.of("ID_8", "subid_8", 8),
            Tuple3.of("ID_9", "subid_9", 9),
            Tuple3.of("ID_10", "subid_10", 10)
            );

        KeyedStream<Tuple3<String, String, Integer>, String> partitionedStream = dataStream.keyBy(new KeySelector<Tuple3<String, String,Integer>, String>() {
                @Override
                public String getKey(Tuple3<String, String, Integer> value) throws Exception {
                    return value.f0; // partition on f0
                }
            });


        partitionedStream.keyBy(new KeySelector<Tuple3<String, String,Integer>, String>() {
                @Override
                public String getKey(Tuple3<String, String, Integer> value) throws Exception {
                    return value.f1; // subid
                }
            }).flatMap(new FilterDuplicate()).print();

        env.execute("Test");
}    
}

---

public  class FilterDuplicate extends RichFlatMapFunction<Tuple3<String, String, Integer>, Tuple3<String, String, Integer>> {

    private ValueState<Boolean> seen;

    @Override
    public void open(Configuration configuration) {
        StateTtlConfig ttlConfig = StateTtlConfig
          .newBuilder(Time.seconds(15))
          .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
          .cleanupFullSnapshot()
          .build();
        ValueStateDescriptor<Boolean> desc = new ValueStateDescriptor<>("seen", Types.BOOLEAN);
        desc.enableTimeToLive(ttlConfig);
        seen = getRuntimeContext().getState(desc);
    }

    @Override
    public void flatMap(Tuple3<String, String, Integer> value, Collector<Tuple3<String, String, Integer>> out) throws Exception {
        
        if (!seen.value()) { // nullpointerexception
            // we haven't seen the element yet
            out.collect(value);
            // set operator state to true so that we don't emit elements with this key again
            seen.update(true);
        }
    }
}

---

  <flink.version>1.17.1</flink.version>
  <target.java.version>11</target.java.version>

NullPointerException

位于

!seen.value()

，采用

flatMap()

方法

为什么

seen.value()

给出NullPointerException？在问题行之前检查了

seen==null

条件，但仍然失败...

Answer 1

小心，否定（！）也会导致问题。 在尝试否定之前，您需要检查两个

seen != null && seen.value() != null

。

看起来你看到的属性没有正确启动，我不理解你的所有代码，但我可能会默认将其设置为 false。

Answer 2

1. 对于

   NullPointerException

   ，只需执行：

    if (seen.value() == null) {
        out.collect(value);
        seen.update(true);
    }

2. 您连续执行了两个

   keyBy()

   操作。当您执行第二个分区时，第一个分区将会丢失。如果您想先按 id 分区，然后按子 id 分区，则应该有一个

   keyBy()

   ，它返回由元组中这两个字段形成的键。