从 Bigquery 导出到存储

Question

将数据从 BigQuery 导出到 Google 存储的最佳方式是什么？请注意，我需要对Bigquery 运行查询，而不是导出所有数据。本质上，我需要对 BigQuery （如 select * from mytable where code=foo ）运行自定义查询，并且查询的结果需要写入存储在 Google Cloud 上的 csv 中。 我相信，最好的方法是通过谷歌数据流。让我知道是否还有其他选择？ 另外，我正在寻找一些有关如何完成此操作的示例。有什么地方可以找到一些例子吗？

这是我目前所拥有的 PipelineOptions pipelineOptions = PipelineOptionsFactory.create(); 管道 p = Pipeline.create(pipelineOptions);

    Date date = new Date();

    p.getOptions().setTempLocation("gs://mybucket/tmp"+date.getTime());

    PCollection<TableRow> rowPCollection = p.apply(BigQueryIO.Read.named("promos")
            .fromQuery("SELECT * FROM [projectid:mydataset.mytable] where id = 256 LIMIT 1000"));

    PCollection<String> stringPCollection = rowPCollection.apply(ParDo.named("Extract").of(new DoFn<TableRow, String>() {
        @Override
        public void processElement(ProcessContext c) {
            TableRow tableRow = c.element();
            try {
                String prettyString = tableRow.toPrettyString();
                c.output(prettyString);
            } catch (IOException e) {
                log.error("Exception occurred:" + e.getMessage());
            }
        }
    }));

    stringPCollection.apply(TextIO.Write.named("WriteOutput").to("gs://mybucket/avexport").withSuffix(".csv"));

    p.run();

运行时，创建 ParDo 时会引发异常

caused by: java.io.NotSerializableException: com.my.validation.CommonValidator
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1184)
at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
at com.google.cloud.dataflow.sdk.util.SerializableUtils.serializeToByteArray(SerializableUtils.java:50)

Answer 1

我猜你的匿名 DoFn 正在从无法序列化的封闭类 (CommonValidator) 中提取一些东西。如果您为 DoFn 实现创建一个静态类，是否可以解决问题？

欲了解更多信息，请参阅NotSerializableException on anonymous class。

Answer 2

除了错误之外，您不必使用 Dataflow 将 BigQuery 数据导出到 GCS，除非您在 Dataflow 管道中进行一些复杂的转换（无论如何，您几乎可以肯定在 SQL/UDF 中进行，但我离题了） .从您的代码 sn-p 和描述来看，您似乎没有对数据进行任何类型的转换。

你可以：

1. 运行 SQL 并将结果保存到 BigQuery 表中。
2. 按照here 的描述将表导出到 GCS。