我有以下情况:
- 管道 A 在 BigQuery 中查找表 A,进行一些计算并返回列名列表。
- 此列名称列表用作管道 B 输出的 BigQuery 架构。
您能否让我知道实现这一目标的最佳选择是什么?
管道 A 可以使用 TextIO 将列名列表写入临时或暂存位置文件,然后管道执行程序读取这些文件以定义管道 B 的架构。如果这种方法看起来不错,请告诉我是否有数据流实用程序,用于从临时或暂存位置读取文件,或者是否应使用 GCS API。
我有以下情况:
您能否让我知道实现这一目标的最佳选择是什么?
管道 A 可以使用 TextIO 将列名列表写入临时或暂存位置文件,然后管道执行程序读取这些文件以定义管道 B 的架构。如果这种方法看起来不错,请告诉我是否有数据流实用程序,用于从临时或暂存位置读取文件,或者是否应使用 GCS API。
您需要执行以下操作:
我不会使用临时位置,因为我们可能会在您开始构建管道 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
最新版本的 Apache Beam 可以做到这一点。请参阅将不同的值写入 Apache Beam 中的不同 BigQuery 表中的自我回答的更一般的问题。