我的场景是此处讨论的场景的一种变体：如何使用数据流执行期间计算的架构写入 BigQuery？ https://stackoverflow.com/questions/29440279/how-do-i-write-to-bigquery-using-a-schema-computed-during-dataflow-execution
在本例中，目标是相同的（在执行期间读取架构，然后将具有该架构的表写入 BigQuery），但我想在单个管道中完成它。
例如，我想将 CSV 文件写入 BigQuery 并避免两次提取该文件（一次读取架构，一次读取数据）。
这可能吗？如果是这样，最好的方法是什么？
我目前最好的猜测是通过侧面输出将架构读入 PCollection，然后使用它来创建表（使用自定义 PTransform），然后再将数据传递给 BigQueryIO.Write。
如果你使用BigQuery.Write https://cloud.google.com/dataflow/java-sdk/JavaDoc/com/google/cloud/dataflow/sdk/io/BigQueryIO.Write要创建表，架构需要知道表的创建时间。
您建议的解决方案是在创建时不指定架构BigQuery.Write https://cloud.google.com/dataflow/java-sdk/JavaDoc/com/google/cloud/dataflow/sdk/io/BigQueryIO.Write转换可能有效，但您可能会收到错误，因为该表不存在，并且您没有配置 BigQueryIO.Write 来在需要时创建它。
您可能需要考虑在主程序中读取足够的 CSV 文件，以便在运行管道之前确定架构。这将避免在运行时确定模式的复杂性。您仍然会承担额外阅读的成本，但希望这是最小的。
或者你创建一个定制水槽 https://cloud.google.com/dataflow/model/sources-and-sinks#creating-sinks将数据写入 BigQuery。你的Sinks https://github.com/GoogleCloudPlatform/DataflowJavaSDK/blob/master/sdk/src/main/java/com/google/cloud/dataflow/sdk/io/Sink.java可以将数据写入GCS。然后，您的 Finalize 方法可以创建一个 BigQueryload job https://cloud.google.com/bigquery/loading-data-into-bigquery。您的自定义接收器可以通过查看记录来推断架构，并使用适当的架构创建 BigQuery 表。
本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)
