如何使用在数据流执行期间计算出的架构写入BigQuery? [英] How do I write to BigQuery using a schema computed during Dataflow execution?

问题描述

我有以下情况:

1. 管道A在BigQuery中查找表A，进行一些计算并返回列名列表.
2. 此列名称列表用作管道B输出的BigQuery架构.

能否让我知道实现这一目标的最佳选择?

管道A是否可以使用TextIO将列名列表写入临时或暂存位置文件，然后由管道执行程序读取以定义管道B的架构.如果这种方法看起来不错，请告诉我是否有一个Dataflow实用程序可以从临时位置或暂存位置或是否应使用GCS API读取文件.

解决方案

您需要执行以下操作:

1. 构造管道A以写入某些位置，例如GCS(在构造管道B时可以引用的任何持久位置都可以使用).
2. 使用BlockingDataflowPipelineRunner运行并等待直到管道A完成.
3. 通过读取您在步骤1中定义的位置来使用架构信息来构建管道B.
4. 运行管道B.

我不会使用临时位置，因为我们可能会在您开始构建管道B之前对其进行清理.可以使用暂存位置(如果与临时位置不同).我还建议使用唯一的文件名，这样，如果管道A多次运行，您就不会在管道B中读入过时的结果.

这应该有助于您读取和写入GCS: https://github.com/GoogleCloudPlatform/DataflowJavaSDK/blob/master/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/GcsUtil.java

您可以从PipelineOptions对象获取GcsUtil的实例: 解决方案

You would need to do the following:

1. Construct Pipeline A to write to some location such as GCS (any durable location which you can reference when constructing pipeline B would work).
2. Use the BlockingDataflowPipelineRunner to run and wait till Pipeline A is done.
3. Construct Pipeline B by using the schema information by reading from the location you defined in step 1.
4. Run Pipeline B.

I would not use the temporary location because we may clean it up before you get around to constructing Pipeline B. The staging location (if different from the temporary location) can be used. I would also advise to use a unique file name so that if Pipeline A runs multiple times, you don't read in stale results with Pipeline B.

This should help you read from and write to GCS: https://github.com/GoogleCloudPlatform/DataflowJavaSDK/blob/master/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/GcsUtil.java

You can get an instance of GcsUtil from the PipelineOptions object: https://github.com/GoogleCloudPlatform/DataflowJavaSDK/blob/master/sdk/src/main/java/com/google/cloud/dataflow/sdk/options/GcsOptions.java#L43

这篇关于如何使用在数据流执行期间计算出的架构写入BigQuery?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！