从Kafka源代码读取时，在光束管道中使用event-time

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions, StandardOptions
from apache_beam.io.kafka import ReadFromKafka
from apache_beam.transforms.window import FixedWindows, TimestampedValue
import datetime

class ParseEvent(beam.DoFn):
    def process(self, element):
        # 假设每条消息是一个 JSON 字符串，包含 'event_time' 字段
        event = json.loads(element[1])
        event_time = datetime.datetime.fromisoformat(event['event_time'])
        yield TimestampedValue(event, event_time.timestamp())

def run():
    options = PipelineOptions()
    options.view_as(StandardOptions).streaming = True

    with beam.Pipeline(options=options) as p:
        events = (
            p
            | 'Read from Kafka' >> ReadFromKafka(consumer_config={'bootstrap.servers': 'localhost:9092'}, topics=['test-topic'])
            | 'Parse events' >> beam.ParDo(ParseEvent())
            | 'Windowing' >> beam.WindowInto(FixedWindows(60))  # 每分钟一个窗口
            | 'Print events' >> beam.Map(print)
        )

if __name__ == '__main__':
    run()