将 CSV 文件从 GCS 导入 BigQuery

Question

我试图弄清楚如何将 CSV 文件从 GCS 加载到 BigQuery。管道如下：

    // Create the pipeline
    Pipeline p = Pipeline.create(options);

    // Create the PCollection from csv
    PCollection<String> lines = p.apply(TextIO.read().from("gs://impression_tst_data/incoming_data.csv"));


    // Transform into TableRow
    PCollection<TableRow> row = lines.apply(ParDo.of(new StringToRowConverter()));


    // Write table to BigQuery
    row.apply(BigQueryIO.<TableRow>writeTableRows()
            .to(“project_id:dataset.table”)
            .withSchema(getSchema())
            .withWriteDisposition(WriteDisposition.WRITE_APPEND)
            .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED));

这是我在 ParDo 中用来创建 TableRow PCollection 的 StringToRowConverter 类：

// StringToRowConverter
static class StringToRowConverter extends DoFn<String, TableRow> {
    @ProcessElement
    public void processElement(ProcessContext c) {
        c.output(new TableRow().set("string_field", c.element()));
    }
}

查看暂存文件，它看起来像这样创建了 JSON 的 TableRows，将 csv 集中到名为“string_field”的单个列中。如果我没有在我的模式中定义 string_field，那么作业就会失败。当我定义 string_field 时，它将 CSV 的每一行写入列，并将架构中定义的所有其他列留空。我知道这是预期的行为。

所以我的问题是：如何获取这个 JSON 输出并将其写入架构？下面的示例输出和架构...

"string_field": "6/26/17 21:28,Dave Smith,1 Learning Drive,867-5309,etc"}

架构：

static TableSchema getSchema() {
            return new TableSchema().setFields(new ArrayList<TableFieldSchema>() {
                // Compose the list of TableFieldSchema from tableSchema.
                {
                    add(new TableFieldSchema().setName("Event_Time").setType("TIMESTAMP"));
                    add(new TableFieldSchema().setName("Name").setType("STRING"));
                    add(new TableFieldSchema().setName("Address").setType("STRING"));
                    add(new TableFieldSchema().setName("Phone").setType("STRING"));
                    add(new TableFieldSchema().setName("etc").setType("STRING"));
                }
            });
        }

有没有比使用 StringToRowConverter 更好的方法呢？

我需要先使用 ParDo 创建一个 TableRow PCollection，然后才能将其写入 BQ。但是，我无法找到一个可靠的示例来说明如何接收 CSV PCollection、转换为 TableRow 并将其写出来。

是的，我是一个想在这里学习的菜鸟。我希望有人可以通过 sn-p 帮助我，或者以最简单的方式为我指明正确的方向。提前致谢。

Answer 1

StringToRowConverterDoFn 中的代码应该解析字符串并生成具有多个字段的TableRow。由于每一行都以逗号分隔，因此这可能涉及将字符串拆分为逗号，然后使用您对列顺序的了解来执行以下操作：

String inputLine = c.element();

// May need to make the line parsing more robust, depending on your
// files. Look at how to parse rows of a CSV using Java.
String[] split = inputLine.split(',');

// Also, you may need to handle errors such as not enough columns, etc.

TableRow output = new TableRow();
output.set("Event_Time", split[0]); // may want to parse the string
output.set("Name", split[1]);
...
c.output(output);