Apache Flink不返回空闲分区的数据

Question

我试图根据事件时间计算Kafka主题每分钟传入事件的速率。为此，我使用了1分钟的TumblingEventTimeWindows。下面给出了代码片段。

我观察到，如果我没有收到特定窗口的任何事件，例如从2.34到2.35，则前一个窗口2.33到2.34不会关闭。我理解在2.33到2.34的窗口中丢失数据的风险（可能是由于系统故障、Kafka滞后更大等原因），但我不能无限期地等待。等待一段时间后，我需要关闭此窗口，系统恢复后，后续窗口可以继续。我怎样才能做到这一点？

StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
    executionEnvironment.setRestartStrategy(RestartStrategies.fixedDelayRestart(
            3,
            org.apache.flink.api.common.time.Time.of(10, TimeUnit.SECONDS)
    ));
    executionEnvironment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
    executionEnvironment.setParallelism(1);
    Properties properties = new Properties();
    properties.setProperty("bootstrap.servers", "localhost:9092");
    properties.setProperty("group.id", "AllEventCountConsumerGroup");
    FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>("event_input_topic", new SimpleStringSchema(), properties);
    DataStreamSource<String> kafkaDataStream = environment.addSource(kafkaConsumer);
    kafkaDataStream
            .flatMap(new EventFlatter())
            .filter(Objects::nonNull)
            .assignTimestampsAndWatermarks(WatermarkStrategy
                    .<Entity>forMonotonousTimestamps()
                    .withIdleness(Duration.ofSeconds(60))
                    .withTimestampAssigner((SerializableTimestampAssigner<Entity>) (element, recordTimestamp) -> element.getTimestamp()))
            .assignTimestampsAndWatermarks(new EntityWatermarkStrategy())
            .keyBy((KeySelector<Entity, String>) Entity::getTenant)
            .window(TumblingEventTimeWindows.of(Time.minutes(1)))
            .allowedLateness(Time.seconds(10))
            .aggregate(new EventCountAggregator())
            .addSink(eventRateProducer);

private static class EntityWatermarkStrategy implements WatermarkStrategy<Entity> {
    @Override
    public WatermarkGenerator<Entity> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {

        return new EntityWatermarkGenerator();
    }

}

private static class EntityWatermarkGenerator implements WatermarkGenerator<Entity> {

    private long maxTimestamp;

    public EntityWatermarkGenerator() {
        this.maxTimestamp = Long.MIN_VALUE + 1;
    }

    @Override
    public void onEvent(Entity event, long eventTimestamp, WatermarkOutput output) {
        maxTimestamp = Math.max(maxTimestamp, eventTimestamp);
    }

    @Override
    public void onPeriodicEmit(WatermarkOutput output) {
        output.emitWatermark(new Watermark(maxTimestamp + 2));
    }
}

此外，我尝试添加一些自定义触发器，但没有帮助。我正在使用Apache Flink 1.11

有人能告诉我，我做错了什么？

当我尝试推送更多具有更新时间戳（例如t1）的主题的数据时，来自更早时间帧（t）的数据会被推送。但是对于t1数据，同样的问题发生在t。

Answer 1

使用flink 1.11 watermarkstrategy api可以帮助您避免抽取虚假数据。您需要的是在每分钟结束时定期生成水印。这是参考：

https://ci.apache.org/projects/flink/flink-docs-stable/dev/event_timestamps_watermarks.html

使用CustomKafkaSerializer创建flinkKafkaConsumer：

Flink Kafka消费者其他消费者=新的Flink Kafka消费者（主题，新的CustomKafkaSerializer（apacheFlink环境加载器），道具）；

如何创建CustomKafkaSerializer？Ans-关于Flink反序列化的两个问题

现在为flinkKafkaConsumer使用水印策略：

FlinkKafkaConsumer<Tuple3<String,String,String>> flinkKafkaConsumer = apacheKafkaConfig.getOtherConsumer();
        flinkKafkaConsumer.assignTimestampsAndWatermarks(new ApacheFlinkWaterMarkStrategy(envConfig.getOutOfOrderDurationSeconds()).
                withIdleness(Duration.ofSeconds(envConfig.getIdlePartitionTimeout())));

这就是水印策略的样子？

Ans-

public class ApacheFlinkWaterMarkStrategy implements WatermarkStrategy<Tuple3<String, String, String>> {

    private long outOfOrderDuration;

    public ApacheFlinkWaterMarkStrategy(long outOfOrderDuration)
    {
        super();
        this.outOfOrderDuration = outOfOrderDuration;
    }

    @Override
    public TimestampAssigner<Tuple3<String, String, String>> createTimestampAssigner(TimestampAssignerSupplier.Context context) {
        return new ApacheFlinkTimeForEvent();
    }

    @Override
    public WatermarkGenerator<Tuple3<String, String, String>> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
        return new ApacheFlinkWaterMarkGenerator(this.outOfOrderDuration);
    } }

这是我们从有效载荷中获取事件时间的方式：

Ans公司-

public class ApacheFlinkTimeForEvent implements SerializableTimestampAssigner<Tuple3<String,String,String>> {

    public static final Logger logger = LoggerFactory.getLogger(ApacheFlinkTimeForEvent.class);

    private static final FhirContext fhirContext = FhirContext.forR4();


    @Override
    public long extractTimestamp(Tuple3<String,String,String> o, long l) {
        //get timestamp from payload
    }
}

这就是我们定期生成水印的方式，这样无论数据是否到达，水印都会在每个分区中每分钟更新一次。

public class ApacheFlinkWaterMarkGenerator implements WatermarkGenerator<Tuple3<String,String,String>> {

    public static final Logger logger = LoggerFactory.getLogger(ApacheFlinkWaterMarkGenerator.class);

    private long outOfOrderGenerator;

    private long maxEventTimeStamp;

    public ApacheFlinkWaterMarkGenerator(long outOfOrderGenerator)
    {
        super();
        this.outOfOrderGenerator = outOfOrderGenerator;
    }

    @Override
    public void onEvent(Tuple3<String, String, String> stringStringStringTuple3, long l, WatermarkOutput watermarkOutput) {
        maxEventTimeStamp = Math.max(maxEventTimeStamp,l);
        Watermark eventWatermark = new Watermark(maxEventTimeStamp);
        watermarkOutput.emitWatermark(eventWatermark);
        logger.info("Current Watermark emitted from event is {}",eventWatermark.getFormattedTimestamp());
    }

    @Override
    public void onPeriodicEmit(WatermarkOutput watermarkOutput) {

        long currentUtcTime = Instant.now().toEpochMilli();
        Watermark periodicWaterMark = new Watermark(currentUtcTime-outOfOrderGenerator);
        watermarkOutput.emitWatermark(periodicWaterMark);
        logger.info("Current Watermark emitted periodically is {}",periodicWaterMark.getFormattedTimestamp());

    }
}

此外，水印的周期性发射必须在应用程序开始时设置。

streamExecutionEnvironment.getConfig().setAutoWatermarkInterval(This is in milliseconds long);

这就是我们如何向flinkKafkaConsumer添加自定义水印和时间戳的方法。

flinkKafkaConsumer.assignTimestampsAndWatermarks(new ApacheFlinkWaterMarkStrategy(Out of Order seconds).
            withIdleness(IdlePartiton Seconds);

Answer 2

有关解决问题的方法，请参阅空闲分区文档。

Answer 3

在您的案例中，withIdleness（）没有帮助的一个原因是，在Kafka源发出数据流之后，您在数据流上调用了assignTimestampsAndWatermarks，而不是在FlinkKafkaConsumer本身上调用它。如果要执行后者，FlinkKafkaConsumer将能够在每个分区的基础上分配时间戳和水印，并将在每个kafka分区的粒度上考虑空闲。有关更多信息，请参阅水印策略和Kafka连接器。

但是，要使此工作正常，您需要使用能够创建带有时间戳的单个流记录的反序列化器，而不是SimpleStringSchema（例如KafkaDeserializationSchema）。有关如何实现KafkaDeserializationSchema的示例，请参阅https://stackoverflow.com/a/62072265/2000823。

但是，请记住，如果所有分区都处于空闲状态，with Idlness（）不会推进水印。它将做的是防止空闲分区阻止水印，如果有来自其他分区的事件，水印可能会推进。