BigQuery writeTableRows始终写入缓冲区问题（Apache Beam+Avro场景）

看起来你遇到的核心问题是批处理与流式处理的行为差异——从Avro文件读取是有界的批处理任务，Pipeline会自动完成并触发BigQuery写入；而Pub/Sub是无界的流式数据源，默认情况下BigQueryIO不会主动触发写入，导致数据一直滞留在缓冲区。下面是具体的排查和解决步骤：

1. 给流式数据添加窗口配置 #

流式数据必须搭配窗口使用，默认的全局窗口（GlobalWindow）不会主动触发输出，只有当Pipeline停止时才会写入。你需要给数据流添加窗口，比如固定时间窗口来控制批次：

p.begin()
 .apply("Input", PubsubIO.readAvros(DataStructure.class).fromTopic("topicName"))
 // 添加1分钟固定窗口，定期归集数据
 .apply("Apply Fixed Window", Window.into(FixedWindows.of(Duration.standardMinutes(1))))
 .apply("Transform", ParDo.of(new CustomTransformFunction()))
 .apply("Load", BigQueryIO.writeTableRows()
     .to(table)
     .withSchema(schema)
     .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND)
     .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED));

2. 配置BigQueryIO的流式触发策略 #

即使加了窗口，你还需要明确BigQueryIO的触发频率，确保即使窗口内数据量不大，也能按时写入：

.apply("Load", BigQueryIO.writeTableRows()
    .to(table)
    .withSchema(schema)
    .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND)
    .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
    // 明确使用流式插入模式
    .withMethod(BigQueryIO.Write.Method.STREAMING_INSERTS)
    // 设置每1分钟触发一次写入，避免数据滞留
    .withTriggeringFrequency(Duration.standardMinutes(1))
    // 可选：允许迟到数据的处理时长
    .withAllowedLateness(Duration.standardHours(1)));

3. 确保水印正常推进 #

如果你的Pub/Sub消息没有携带有效时间戳，或者时间戳是过去很久的值，Beam的水印无法正常推进，窗口也不会关闭。可以手动给消息分配时间戳：

方式一：从Pub/Sub消息属性读取时间戳 #

如果消息带有event_timestamp这类属性，可以直接指定：

.apply("Input", PubsubIO.readAvros(DataStructure.class)
    .fromTopic("topicName")
    .withTimestampAttribute("event_timestamp"))

方式二：自定义分配当前时间作为时间戳 #

如果消息没有时间戳属性，用DoFn手动分配：

.apply("Input", PubsubIO.readAvros(DataStructure.class).fromTopic("topicName"))
.apply("Assign Timestamp", ParDo.of(new DoFn<DataStructure, DataStructure>() {
    @ProcessElement
    public void processElement(ProcessContext ctx) {
        // 用当前时间作为消息时间戳，推进水印
        ctx.outputWithTimestamp(ctx.element(), Instant.now());
    }
}))

4. 验证数据流是否正常输出 #

先排除自定义转换函数的问题，在转换后添加日志，确认数据是否正常流向BigQuery：

.apply("Transform", ParDo.of(new CustomTransformFunction()))
.apply("Log Rows", LogElements.via((TableRow row) -> "Preparing to write: " + row))
.apply("Load", ...)

5. 检查Pipeline运行模式 #

如果用DirectRunner测试，必须添加--streaming=true参数，否则会按批处理模式运行，无法处理无界的Pub/Sub数据；如果是DataflowRunner，它会自动识别流式数据源，但也可以明确指定--streaming=true。

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