Apache Beam消费Kafka Topic异常：无输出文件生成问题问询

1. 违背Beam执行模型：你在beam.Map里直接用open写本地文件的方式，不符合Apache Beam的设计逻辑。Beam的转换操作基于分布式并行执行，流式模式下数据按窗口批量处理，手动写文件无法适配Beam的延迟执行与触发机制，导致数据不会被实际写入。
2. 流式触发逻辑缺失：流式模式下，Beam需要明确的窗口和触发配置来输出数据。当前代码未设置窗口，Beam无法判断何时输出处理后的数据，即便有消息进入也不会执行写入。而添加max_num_records=1后，作业转为有限流处理（类批处理），处理完指定消息后立即终止，所有操作被强制执行，因此文件会被创建。

使用Beam官方的WriteToText连接器实现流式写入，它适配Beam执行模型，支持流式模式下的持续输出。

修改后的代码示例：

import json
from typing import Optional, List
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions, StandardOptions

def run(beam_args: Optional[List[str]] = None) -> None:
    TOPIC = "topic"
    OUTPUT_FILE = "output.json"
    beam_options = PipelineOptions(beam_args, save_main_session=True)
    beam_options.view_as(StandardOptions).streaming = True
    
    # 配置消费者，确保能读取历史或新消息
    CONSUMER_CONFIG = {
        'bootstrap.servers': 'localhost:9092',
        'group.id': 'beam-consumer-group',
        'auto.offset.reset': 'earliest'
    }

    with beam.Pipeline(options=beam_options) as pipeline:
        msg_kv_bytes = (
            pipeline 
            | 'ReadData' >> beam.io.ReadFromKafka(
                consumer_config=CONSUMER_CONFIG,
                topics=[TOPIC],
            )
        )

        # 提取消息内容并格式化为JSON字符串
        formatted_messages = (
            msg_kv_bytes
            | 'ExtractAndFormat' >> beam.Map(lambda msg: json.dumps(json.loads(msg[1])) + '\n')
        )

        # 用WriteToText实现流式写入
        _ = formatted_messages | 'WriteToFile' >> beam.io.WriteToText(
            file_path_prefix=OUTPUT_FILE,
            file_name_suffix='',
            append=True,
            num_shards=1  # 本地测试设为1，避免生成多分片文件
        )

if __name__ == "__main__":
    run()

• WriteToText会自动处理流式模式下的窗口与触发逻辑，持续将新消息写入文件。
• num_shards=1适合本地测试，避免生成多个分片文件；生产环境可按需调整分片数量。
• auto.offset.reset设为earliest时，作业启动后会消费历史未处理消息；若只需消费启动后的新消息，可改为latest。

内容的提问来源于stack exchange，提问作者Dulanga Heshan