在pyflink中访问kafka时间戳

Question

我正在尝试编写一个用于测量延迟和吞吐量的Pyflink应用程序。我的数据作为来自kafka主题的json对象，并使用SimpleStringSchema类加载到Datastream中进行反序列化。下面是对这篇文章的回答（如何在Kafka和Flink环境中测试性能？）我让Kafka的制作人在事件中加上时间戳，但我很难理解我现在如何才能访问这些时间戳。我知道上面提到的文章为这个问题提供了一个解决方案，但是我很难将这个例子转移到python，因为文档/例子很少。

class MyProcessFunction():

    def process_element(self, value, ctx):
        result = value.get_time_stamp()
        yield result

在此处执行value.get_time_stamp()的正确方法是什么？或者有没有一个更简单的方法来解决我不知道的问题？

谢谢！

Answer 1

设置由Kafka主题支持的表时，可以为Kafka时间戳声明一个虚拟列，如本例中的event_time列：

CREATE TABLE KafkaTable (
  `event_time` TIMESTAMP(3) METADATA FROM 'timestamp',
  `partition` BIGINT METADATA VIRTUAL,
  `offset` BIGINT METADATA VIRTUAL,
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING
) WITH (
  'connector' = 'kafka',
  'topic' = 'user_behavior',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'csv'
);

有关使用Kafka头中的元数据的更多信息，请参阅Flink的Kafka表连接器的文档。