如何将 Pcollection<String> 变量转换为字符串

Posted 2023-03-25

【中文标题】如何将 Pcollection<String> 变量转换为字符串

【英文标题】：How To convert Pcollection<String> variable into String

【发布时间】：2018-03-06 12:31:55

【问题描述】：

我有 PCollection 类型的字符串，我想对其进行转换以从 BigQuery 表中获取特定列的值。所以我使用 BigQueryIO.readTableRows 从 BigQuery 获取值。

这是我的代码：

     PCollection<TableRow> getConfigTable = pipeline.apply("read from Table",
            BigQueryIO.readTableRows().from("TableName"));

    RetrieveDestTableName retrieveDestTableName = new RetrieveDestTableName();
    PCollection<String> getDestTableName = getConfigTable.apply(ParDo.of(new DoFn<String,String>()
        @ProcessElement
        public void processElement(ProcessContext c)
            c.output(c.element().get("ColoumnName").toString());
        
    ));

根据上面的代码，我将从 getDestTableName 获得 PCollection 类型的输出，但我希望在 String 变量中获得此输出。

有什么方法可以将 PCollection 转换为 String 数据类型变量，以便我能够在我的代码中使用变量？

【参考方案1】：

在 Apache Beam 编程模型中无法将 PCollection<String> 转换为 String。 PCollection 简单地描述了管道在任何给定点的状态。在开发过程中，您无法直接访问 PCollection 中的字符串。

您可以通过转换处理PCollection 中的字符串。但是，您似乎需要表配置来构建管道的其余部分。您需要提前知道目的地，或者您可以使用DynamicDestinations 来确定在管道执行期间要写入哪个表。您无法从PCollection 中获取表配置值并使用它来进一步构建管道。

【讨论】：

感谢@Andrew，它对我有用。是否可以在 DynamicDestination 转换中动态获取 BigQuery Table 架构信息？你能用例子解释一下吗，这对我很有帮助。

@ShubhamR，DynamicDestinations 转换不提供动态检索TableSchema 的方法。理想情况下，您希望提前将所有TableSchemas 提供给转换，以便在执行期间（开销）不检索架构。在getSchema() 方法中，您可以实现一些简单的东西，例如一个if 语句，它根据destination: String 参数确定要使用哪个TableSchema。

【参考方案2】：

您似乎想要 JdbcIO.readAll() 之类的东西，但对于 BigQuery，允许管道动态计算读取配置。目前尚未为 BigQuery 实现此功能，但这是一个合理的要求。

同时您的选择是：

用更复杂的 BigQuery SQL 查询表达您正在执行的操作，并使用单个 BigQueryIO.read().fromQuery() 表达您在不使用 Beam API 的情况下提取感兴趣的表的管道部分，而是直接使用 BigQuery API，因此您正在操作常规 Java 变量而不是 PCollections。