Flink - 如何使用 withTimestampAssigner 从事件有效负载中获取时间（不使用 Kafka 时间戳）

Question

我试图了解如何在 Kafka Source 的 WatermarkStrategy 中使用 withTimestampAssigner()。我需要使用的“时间”在消息负载中。

为此，我有以下代码：

FlinkKafkaConsumer<Event> kafkaData =
        new FlinkKafkaConsumer("CorID_0", new EventDeserializationSchema(), p);
kafkaData.assignTimestampsAndWatermarks(
        WatermarkStrategy
        .forMonotonousTimestamps()
                .withTimestampAssigner(Event, Event.time))

DataStream<Event> stream = env.addSource(kafkaData);

EventDeserializationSchema() 是这样的：

public class EventDeserializationSchema implements DeserializationSchema<Event> {

    private static final long serialVersionUID = 1L;
    
    private static final CsvSchema schema = CsvSchema.builder()
            .addColumn("firstName")
            .addColumn("lastName")
            .addColumn("age", CsvSchema.ColumnType.NUMBER)
            .addColumn("time")
            .build();

    private static final ObjectMapper mapper = new CsvMapper();

    @Override
    public Event deserialize(byte[] message) throws IOException {
        return mapper.readerFor(Event.class).with(schema).readValue(message);
    }

    @Override
    public boolean isEndOfStream(Event nextElement) {
        return false;
    }

    @Override
    public TypeInformation<Event> getProducedType() {
        
        return TypeInformation.of(Event.class);
    }
}

还有事件：

import java.io.Serializable;

public class Event implements Serializable {
    public String firstName;
    public String lastName;
    private int age;
    public String time;

    public Event() {
    }

    public String getFirstName() {
        return firstName;
    }

    public void setFirstName(String firstName) {
        this.firstName = firstName;
    }

    public String getLastName() {
        return lastName;
    }

    public void setLastName(String lastName) {
        this.lastName = lastName;
    }

    public int getAge() {
        return age;
    }

    public void setAge(int age) {
        this.age = age;
    }

    public String getTime() {
        return time;
    }

    public void setTime(String time) {
        this.time = time;
    }
}

我想了解的是如何为 withTimeStampAssigner() 提供 时间：

.withTimestampAssigner(???))

变量应该是 Event.time 但从 flink 页面我不太明白。

我一直在寻找

这让我有点困惑，因为我不明白在我的情况下解决方案是否非常简单，或者我需要额外的上下文。我发现的所有示例都使用 .forBoundedOutOfOrderness() 或以前版本的 flink，其中实现与此不同：

kafka flink timestamp Event time and watermark

谢谢！

Answer 1

如果源（例如，FlinkKafkaConsumer）没有提供您想要使用的时间戳，那么您需要提供一个TimestampAssigner。这是一个将事件和先前分配的时间戳（如果有）作为输入并返回时间戳的函数。在您的情况下，可能看起来像这样：

FlinkKafkaConsumer<Event> kafkaData =
        new FlinkKafkaConsumer("CorID_0", new EventDeserializationSchema(), p);

WatermarkStrategy<Event> wmStrategy = 
        WatermarkStrategy
          .<Event>forMonotonousTimestamps()
          .withTimestampAssigner((event, timestamp) -> event.getTime());

DataStream<Event> stream = env.addSource(
        kafkaData.assignTimestampsAndWatermarks(wmStrategy));

（注意：这不太可行，因为您的 getTime() 方法返回一个字符串。您需要解析字符串并返回一个 long ——通常它是一个 long 表示自纪元以来的毫秒数。）

涉及TimestampAssignerSupplier.Context 或WatermarkGeneratorSupplier.Context 的情况适用于您需要访问较低级别的API 以执行更多自定义操作的情况。在这种情况下，这不是必需的。