结合使用数据流和BigQueryIO.Write.Method.STORAGE_WRITE_API的示例

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.values.PCollection;

public class DataflowExample {
  public static void main(String[] args) {
    // 创建数据流管道
    PipelineOptions options = PipelineOptionsFactory.create();
    Pipeline pipeline = Pipeline.create(options);

    // 创建数据集
    PCollection<String> data = pipeline.apply(Create.of("data1", "data2", "data3"));

    // 将数据写入BigQuery
    data.apply(
        BigQueryIO.writeTableRows()
            .to("project_id:dataset.table")
            .withMethod(BigQueryIO.Write.Method.STORAGE_WRITE_API)
            .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
            .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND));

    // 运行数据流管道
    pipeline.run();
  }
}