4

我有以下情况:

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

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

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

4

2 回答 2

5

您需要执行以下操作:

  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 实例: https://github.com/GoogleCloudPlatform/DataflowJavaSDK/blob/master/sdk/src/main/java/com/google/cloud/dataflow/sdk/options/GcsOptions。爪哇#L43

于 2015-04-03T22:01:41.810 回答
1

最新版本的 Apache Beam 可以做到这一点。请参阅将不同的值写入 Apache Beam 中的不同 BigQuery 表中的自我回答的更一般的问题。

于 2017-05-04T00:30:34.897 回答