空闲源计时器上的Flink Keyedprocess函数

Question

我有一个听Kafka的Flink过程。然后，将消耗的消息保存在并发哈希映射中一段时间，然后需要将其放入cassandra。

操作员链类似于

DataStream<Message> datastream = KafkaSource.createsource();
DataStream<Message> decodededMessage = datastream.flatmap(new DecodeMessage());

decodedMessage.assigneTimestampsandWatermarks(new AscendingTimestampExtractor<Message>(){

  public long extractAscendingTimestamp(Message m){

      return message.getTimestamp();
  }
  
}).keyBy((KeySelector<Message>) x-> x.getID())
  .process(new Timerfunction())
  .addSink(new MySink());



class TimerFunction extends KeyedProcessFunction<Integer,Message,Message>{

    private ValueState<Message> x;

    public void processElement(){
//some logic to create timestamp for one minute

     context.timerService().registerEventTimeTimer(x.getTimestamp());
     }
  
    public void onTimer()
  // output values on trigger
}

在使用EventTime时，我有一些疑问

• 消息将具有唯一的id、时间戳和其他一些属性。一分钟内可能有一百万个唯一的钥匙。keyBy操作会影响性能吗

我需要涵盖以下场景

>

ID为1的X消息在8小时1分1秒到达

ID为2的Y消息在8小时1分4秒到达

由于我使用Id作为密钥，这两条消息都应该设置一个计时器，以8小时2分钟0秒触发。根据闪烁留档，如果计时器的时间戳相同，它将只触发一次。我面临一个问题，源空闲几分钟，计时器一直在等待下一个水印，永远不会触发。如何处理空闲源？

在这种情况下，使用processingtime是更好的选择吗？

我也有使用Flink v1的限制。8所以需要一些关于该版本的信息。

提前感谢

Answer 1

我不完全理解你的问题；缺少太多背景。但是我可以提供几点：

（1） keyBy代价高昂：它强制序列化/反序列化以及网络洗牌。

（2） 当且仅当计时器用于相同的时间戳和相同的密钥时，才会对其进行重复数据消除。

（3） 至于空闲源问题，当事件再次开始流动时，事件时间计时器将最终触发，因为这将推进水印。如果迫不及待，可以使用https://github.com/aljoscha/flink/blob/6e4419e550caa0e5b162bc0d2ccc43f6b0b3860f/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/timestamps/ProcessingTimeTrailingBoundedOutOfOrdernessTimestampExtractor.java，或切换到处理时间。