Apache Beam Python按30秒窗口将Kafka数据写入GCS生成独立Parquet文件问题

一、按30秒窗口生成单个Parquet文件修复 #

问题根因 #

• beam.Map会对窗口内的每一条消息单独执行一次写入逻辑，自然每条消息生成一个文件
• 直接在转换函数中使用datetime.now()生成文件名无法对应窗口时间，且全局定义的folder_name仅在管道启动时生成一次，不会随日期自动更新
• 你自定义的trigger配置冗余，反而干扰窗口触发逻辑

修复后代码 #

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from beam_nuggets.io import kafkaio
import json
from datetime import datetime
import pandas as pd
import config as conf
import apache_beam.transforms.window as window
from apache_beam import DoFn, ParDo

consumer_config = {
    "topic": "Uswrite",
    "bootstrap_servers": "*.*.*.*:9092",
    "group_id": "notification_consumer_group_33",
    # 注意kafka-python的配置用下划线，不是点，这里是offset不生效的核心原因
    "auto_offset_reset": "earliest",
    "enable_auto_commit": True,
    "auto_commit_interval_ms": 5000
}

class WriteWindowToParquet(DoFn):
    def process(self, element, window=beam.DoFn.WindowParam):
        # 用窗口开始时间生成文件名和目录，避免时间错乱
        window_start = datetime.fromtimestamp(window.start)
        folder_name = window_start.strftime('%Y-%m-%d')
        file_name = window_start.strftime("%Y_%m_%d-%H_%M_%S")
        # element是当前窗口所有消息的列表
        data_list = [json.loads(msg[1]) for msg in element]
        df = pd.DataFrame(data_list)
        # 写入GCS
        df.to_parquet(
            f'gs://{conf.gcs}/{folder_name}/{file_name}.parquet',
            storage_options={"token": "gcp.json"}, 
            engine='fastparquet'
        )
        yield f"写入完成：{file_name}.parquet，共{len(data_list)}条数据"

with beam.Pipeline(options=PipelineOptions()) as p:
    _ = (p 
        | "Reading messages from Kafka" >> kafkaio.KafkaConsume(consumer_config=consumer_config)
        # 移除多余的trigger配置，FixedWindows默认触发逻辑即可满足30秒窗口需求
        | 'Windowing' >> beam.WindowInto(window.FixedWindows(30), allowed_lateness=0)
        # 把窗口内所有消息聚合成一个列表，每个窗口仅输出一个列表元素
        | 'Aggregate window data' >> beam.CombineGlobally(beam.combiners.ToListCombiner()).without_defaults()
        | 'Write to parquet' >> ParDo(WriteWindowToParquet())
    )

二、Kafka offset不从最早位置消费的解决方案 #

核心原因是你用的beam_nuggets.io.kafkaio底层依赖kafka-python库，该库的配置参数用下划线而非点分隔，你配置的"auto.offset.reset": "earliest"格式不被识别，因此使用默认的latest策略，修改为"auto_offset_reset": "earliest"即可生效。若仍不生效，可删除旧消费者组的已提交offset后重启管道。

三、trigger、allowed_lateness、accumulation_mode使用说明 #

• trigger：固定30秒处理时间窗口场景下无需自定义trigger，FixedWindows的默认触发逻辑就是窗口时间到了就触发一次，你之前加的AfterProcessingTime(30)属于冗余配置，反而可能导致重复触发
• allowed_lateness：该参数用于处理事件时间场景下的迟到数据，你使用的是处理时间窗口，不存在迟到数据，直接设为0即可，无需设置900秒
• accumulation_mode：该参数仅在窗口需要多次触发（比如每隔5秒触发一次输出当前窗口累计结果）时需要配置，你当前场景每个窗口仅触发一次，无需额外配置，默认即可

内容的提问来源于stack exchange，提问作者Emsal Cengiz