Apache Flink流事件延迟

Question

我试图开发以下代码，但它不起作用。我想使用apache Flink来延迟时间（在时间戳字段中指定的）与当前日期不同的事件。

样品：

>

当前日期：2022-05-06 10:30

事件1[{“user1”：“1”，“user2”：“2”，“timestamp”：“2022-05-06 10:30”}--

事件2[{“user1”：“1”，“user2”：“2”，“timestamp”：“2022-05-06 13:30”}--

FLinkLikePipeline

public class FlinkLikePipeline {

public static void createBackup() throws Exception {
    String inputTopic = "flink_input";
    String outputTopic = "flink_output";
    String consumerGroup = "baeldung";
    String kafkaAddress = "localhost:9092";

    StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
    environment.getConfig().setAutoWatermarkInterval(Duration.ofMillis(1).toMillis());

    FlinkKafkaConsumer011<Like> consumer = Consumers.createMsgLike(inputTopic, kafkaAddress, consumerGroup);
    consumer.assignTimestampsAndWatermarks(new InputMessageTimestampAssigner());

    FlinkKafkaProducer011<Matches> producer = new FlinkKafkaProducer011<Matches>(kafkaAddress, outputTopic,
            new BackupSerializationSchema());

    DataStream<Like> likes = environment.addSource(consumer);

    DataStream<Matches> matches = likes.keyBy(like -> like.getId()).process(new MatchFunction());
    matches.addSink(producer);

    environment.execute("Keyed Process Function Example");

}

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

}

InputMessageTimestampAssigner

public class InputMessageTimestampAssigner implements 
AssignerWithPunctuatedWatermarks<Like> {

@Override
public long extractTimestamp(Like element, long previousElementTimestamp) {
    Timestamp now = new Timestamp(System.currentTimeMillis());
    return now.getTime();

}

@Override
public Watermark checkAndGetNextWatermark(Like lastElement, long extractedTimestamp) {
    return new Watermark(extractedTimestamp);
}

}

匹配函数

public class MatchFunction extends KeyedProcessFunction<String, Like, Matches> {

ValueState<Like> liked;

@Override
public void open(Configuration parameters) throws Exception {
    liked = getRuntimeContext().getState(new ValueStateDescriptor<>("like", Like.class));
}

@Override
public void processElement(Like newLike, Context ctx, Collector<Matches> out) throws Exception {

    Timestamp now = new Timestamp(System.currentTimeMillis());
    Timestamp sendAt = newLike.getTimestamp();

    if (sendAt.before(now)) {
        liked.clear();
        out.collect(new Matches(newLike.getUser1(), newLike.getUser2(), newLike.getTimestamp()));
    } else {
        // schedule the next timer by sendAt
        liked.update(newLike);
        ctx.timerService().registerEventTimeTimer(sendAt.getTime());
    }
}

@Override
public void onTimer(long timestamp, OnTimerContext ctx, Collector<Matches> out) throws Exception {
    Like like = liked.value();
    Timestamp now = new Timestamp(System.currentTimeMillis());
    Timestamp sendAt = like.getTimestamp();
    if (now == sendAt) {
        liked.clear();
        out.collect(new Matches(like.getUser1(), like.getUser2(), like.getTimestamp()));
    }

}

}

制作人Topic Kafka（Json Sample）：{"user1"："1"，"user2"："2"，"时间戳"："2022-05-06T11:32:00.000Z"}

谢谢你的支持，朱塞佩。

Answer 1

尝试以下操作：

public class MatchFunction extends KeyedProcessFunction<String, Like, Matches> {

ValueState<Like> liked;

@Override
public void open(Configuration parameters) throws Exception {
    liked = getRuntimeContext().getState(new ValueStateDescriptor<>("like", Like.class));
}

@Override
public void processElement(Like newLike, Context ctx, Collector<Matches> out) throws Exception {

    Timestamp now = new Timestamp(ctx.timestamp());  // CHANGE: Use context timestamp
    Timestamp sendAt = newLike.getTimestamp();

    if (sendAt.before(now)) {
        liked.clear();
        out.collect(new Matches(newLike.getUser1(), newLike.getUser2(), newLike.getTimestamp()));
    } else {
        // schedule the next timer by sendAt
        liked.update(newLike);
        ctx.timerService().registerProcessingTimeTimer(sendAt.getTime()); // CHANGED: change to processing time
    }
}

@Override
public void onTimer(long timestamp, OnTimerContext ctx, Collector<Matches> out) throws Exception {
    Like like = liked.value();
    Timestamp now = new Timestamp(timestamp);  // CHANGE: Use current timestamp
    Timestamp sendAt = like.getTimestamp();
    if (now >= sendAt) { // CHANGE: Should try equals or great
        liked.clear();
        out.collect(new Matches(like.getUser1(), like.getUser2(), like.getTimestamp()));
    }

}

和

public static void createBackup() throws Exception {
    String inputTopic = "flink_input";
    String outputTopic = "flink_output";
    String consumerGroup = "baeldung";
    String kafkaAddress = "localhost:9092";

    StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
    environment.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);  // CHANGED: to processingtime
    environment.getConfig().setAutoWatermarkInterval(Duration.ofMillis(1).toMillis());

    FlinkKafkaConsumer011<Like> consumer = Consumers.createMsgLike(inputTopic, kafkaAddress, consumerGroup);
    consumer.assignTimestampsAndWatermarks(new InputMessageTimestampAssigner());

    FlinkKafkaProducer011<Matches> producer = new FlinkKafkaProducer011<Matches>(kafkaAddress, outputTopic,
            new BackupSerializationSchema());

    DataStream<Like> likes = environment.addSource(consumer);

    DataStream<Matches> matches = likes.keyBy(like -> like.getId()).process(new MatchFunction());
    matches.addSink(producer);

    environment.execute("Keyed Process Function Example");

}