如何在Flink中打印聚合数据流？

Question

我有一个表示为set 的自定义状态计算，当我的数据流 > 看到来自Kafka的新事件时，它将不断更新。现在，每次更新状态时，我都希望将更新后的状态打印到stdout。想知道怎么在Flink中做到这一点吗？与所有的窗口和触发器操作很少混淆，我一直得到以下错误。

原因：java.lang.RuntimeException：记录具有long.min_value时间戳(=没有时间戳标记）。时间特性设置为“ProcessingTime”，还是忘记调用“DataStream.AssignTimeStampsandWatermarks（...）”？

我只想知道如何将我的聚合流Datastream > 打印到stdout或写回另一个kafka主题？

下面是引发错误的代码片段。

StreamTableEnvironment bsTableEnv = StreamTableEnvironment.create(env, bsSettings);

DataStream<Set<Long>> stream = bsTableEnv.toAppendStream(kafkaSourceTable, Row.class)
   stream
      .aggregate(new MyCustomAggregation(100))
      .process(new ProcessFunction<Set<Long>, Object>() {
       @Override
         public void processElement(Set<Long> value, Context ctx, Collector<Object> out) throws Exception {
           System.out.println(value.toString());
         }
       });

Answer 1

使用Flink将集合保持在状态可能非常昂贵，因为在某些情况下，集合将频繁地序列化和反序列化。如果可能，则首选使用Flink的内置ListState和MapState类型。

下面是一个示例,说明了以下几点：

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

    StreamExecutionEnvironment env =
        StreamExecutionEnvironment.getExecutionEnvironment();

    env.fromElements(1L, 2L, 3L, 4L, 3L, 2L, 1L, 0L)
        .keyBy(x -> 1)
        .process(new KeyedProcessFunction<Integer, Long, List<Long>> () {
            private transient MapState<Long, Boolean> set;

            @Override
            public void open(Configuration parameters) throws Exception {
                set = getRuntimeContext().getMapState(new MapStateDescriptor<>("set", Long.class, Boolean.class));
            }

            @Override
            public void processElement(Long x, Context context, Collector<List<Long>> out) throws Exception {
                if (set.contains(x)) {
                    System.out.println("set contains " + x);
                } else {
                    set.put(x, true);
                    List<Long> list = new ArrayList<>();
                    Iterator<Long> iter = set.keys().iterator();
                    iter.forEachRemaining(list::add);
                    out.collect(list);
                }
            }
        })
        .print();

    env.execute();

}

请注意，我希望使用键控状态，但事件中没有任何可用作键的内容，所以我只是通过一个常量对流进行键控。这通常不是一个好主意，因为它阻止了并行处理--但是由于您是作为一个集合进行聚合，所以不能并行处理，所以没有坏处。

[1]
[1, 2]
[1, 2, 3]
[1, 2, 3, 4]
set contains 3
set contains 2
set contains 1
[0, 1, 2, 3, 4]