Google Dataflow：PCollection<String> 到 PCollection<TableRow> 用于 BigQuery 插入

Question

我对 Google Cloud Platform 还很陌生，我是第一次尝试 Google Dataflow 来为我的研究生课程做一个项目。我想做的是编写一个自动加载作业，从我的 Cloud Storage 上的某个存储桶加载文件，并将其中的数据插入到 BigQuery 表中。

我将数据作为 PCollection<String> 类型获取，但为了插入 BigQuery，我显然需要将其转换为 PCollection<TableRow> 类型。到目前为止，我还没有找到可靠的答案。

这是我的代码：

public static void main(String[] args) {
    //Defining the schema of the BigQuery table
    List<TableFieldSchema> fields = new ArrayList<>();
    fields.add(new TableFieldSchema().setName("Datetime").setType("TIMESTAMP"));
    fields.add(new TableFieldSchema().setName("Consumption").setType("FLOAT"));
    fields.add(new TableFieldSchema().setName("MeterID").setType("STRING"));
    TableSchema schema = new TableSchema().setFields(fields);

    //Creating the pipeline
    PipelineOptions options = PipelineOptionsFactory.fromArgs(args).withValidation().create();
    Pipeline p = Pipeline.create(options);

    //Getting the data from cloud storage
    PCollection<String> lines = p.apply(TextIO.Read.named("ReadCSVFromCloudStorage").from("gs://mybucket/myfolder/certainCSVfile.csv"));

    //Probably need to do some transform here ...

    //Inserting data into BigQuery
    lines.apply(BigQueryIO.Write
            .named("WriteToBigQuery")
            .to("projectID:datasetID:tableID")
            .withSchema(schema)
            .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND)
            .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED));
}

我可能只是忘记了一些基本的东西，所以我希望你们能帮助我......

Answer 1

BigQueryIO.Write 在PCollection<TableRow> 上运行，如Writing to BigQuery 中所述。您需要应用转换将PCollection<TableRow> 转换为PCollection<String>。举个例子，看看StringToRowConverter：

  static class StringToRowConverter extends DoFn<String, TableRow> {
    /**
     * In this example, put the whole string into single BigQuery field.
     */
    @Override
    public void processElement(ProcessContext c) {
      c.output(new TableRow().set("string_field", c.element()));
    }
  ...
}