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

Question

我试图理解如何在Kafka源代码的水印策略中使用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页面我不太明白。

我一直在寻找

这让我有点困惑，因为我不明白在我的案例中，解决方案是非常直接的，还是我需要额外的背景。我找到的所有例子都是一样的。ForBoundedAutoforderness（）或flink的早期版本，其实现与以下版本不同：

Kafka·Flink时间戳事件时间和水印

谢啦！

Answer 1

如果源（例如，Flink Kafka消费者）没有提供您要使用的时间戳，那么您需要提供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，表示从epoch开始的毫秒数。）

涉及时间分配供应商的案例。上下文或水印生成器供应商。上下文适用于需要访问较低级别API才能执行更多自定义操作的情况。在这种情况下没有必要这样做。