从 Bigquery 导出到存储

Posted 2023-03-24

【中文标题】从 Bigquery 导出到存储

【英文标题】：export from Bigquery to Storage

【发布时间】：2017-01-05 00:15:23

【问题描述】：

将数据从 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)

【参考方案1】：

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

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

【参考方案2】：

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

你可以：

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

【讨论】：

感谢您的建议。我拥有的数据在谷歌云数据存储中，数据也在大数据中，所以数据在两个地方。所以选项是选项 1： 1. 对数据存储运行 SQL 查询并写入 bigQuery。 2. 然后从 BigQuery 导出到存储选项 2： 1. 对 BigQuery 运行 SQL 查询并写入 BigQuery 中的另一个表 2. 然后从 BigQuery 导出到存储选项 2.1 在没有数据流的情况下是否可行？

@verma - 您在问题中从未提及有关 Cloud Datastore 的任何内容。其次，Cloud Datastore 是一种 NoSQL 解决方案，因此您将无法“针对数据存储运行 SQL 查询并写入 bigQuery”。

是的。我们在这两个地方都有数据写入。云存储是我们的主要数据库，我们在 BigQuery 中复制数据只是为了这个用例。所以根据你说的，这就是我的想法。 1. 对主数据库 (Cloud Datastore) 执行查询 2. 将 BigQuery 中的数据写入新表 'mytable-uuid' 3. 从步骤 2 中创建的表执行导出到 Cloud-Storage 我应该怎么做执行所有这些步骤？ Cloud-DataFlow 不是在这里使用的最佳工具吗？

有没有办法可以将数据作为 csv 从数据存储导出到存储？本质上，我想对数据存储运行查询，然后将结果导出为 csv？