如何在Java DataFlow管道中将Apache Beam Row转为BigQuery TableRow？

你完全不需要自己编写DoFn来转换Row到TableRow，Apache Beam已经提供了原生支持，可直接处理带Schema的PCollection<Row>写入BigQuery，包括ARRAY(INT64)这类复杂类型的自动转换。

核心解决方案：使用BigQueryIO.writeRows() #

BigQueryIO.writeRows()是专门为Beam Schema的Row类型设计的写入API，内部已封装所有类型映射逻辑（包括数组、嵌套结构、不同数值类型的转换），无需手动处理。

替换后的代码示例 #

// 若需指定BigQuery表结构，可保留原tableSchema；也可让Beam自动从Row的Schema推断
TableSchema tableSchema = // omitted

BigQueryIO.Write<Row> bigQuerySink = BigQueryIO.writeRows()
    .to(new TableReference()
        .setProjectId(options.getProject())
        .setDatasetId(options.getBigQueryDataset())
        .setTableId(options.getBigQueryTable()))
    // 可选：如需自定义表结构则传入tableSchema；否则可省略，Beam会自动推断
    .withSchema(tableSchema)
    .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
    .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND);

关键说明 #

• 类型自动映射：Beam会自动将Beam Schema类型转换为对应BigQuery类型，例如：
  • Beam的ArrayType.of(SqlTypeName.BIGINT) → BigQuery的ARRAY<INT64>
  • 嵌套Row → BigQuery的STRUCT类型
  • 字符串、布尔值、浮点等基础类型也会自动对齐
• Schema推断：如果PCollection<Row>已附带Beam Schema，可省略withSchema(tableSchema)，Beam会自动根据Row的Schema生成对应BigQuery表结构（前提是表不存在且CREATE_IF_NEEDED生效）
• 环境兼容：该API支持DataFlow所有运行环境，包括批处理和流处理场景

若坚持使用原write()方法的替代方案 #

如果因特殊原因必须使用基于TableRow的BigQueryIO.write()，可以直接使用Beam提供的RowToTableRow转换工具（位于org.apache.beam.sdk.io.gcp.bigquery包下）：

import org.apache.beam.sdk.io.gcp.bigquery.RowToTableRow;

PCollection<TableRow> tableRowCollection = rowCollection.apply(ParDo.of(new RowToTableRow()));

不过更推荐使用writeRows()，它是更简洁、原生的解决方案。

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