如何将运行时查询传递给 apache beam 中的 SqlTransform？

Question

我想在运行数据流作业时动态传递查询。我正在使用 SQLTransform，当我在代码中传递 Query 时它工作正常。

我的用例需要在运行时传递查询，是否可以在 Apache Beam 中使用 SqlTransform？

如果我在代码中对其进行硬编码，这将起作用。

String PQuery = "SELECT col1, max(col2) as max_watermark FROM PCOLLECTION GROUP BY col1";
PCollection<Row> rows1 =   rows.apply(SqlTransform.query(PQuery));

但是对于valueProvider 输入，它会给出编译时错误。

PCollection<Row> rows1 =   rows.apply(SqlTransform.query(options.getQuery()))

错误

The method query(String) in the type SqlTransform is not applicable for the arguments (ValueProvider<String>)

Answer 1

要解决您的问题，您需要获取 ValueProvider 中的值：

PCollection<Row> rows1 = rows.apply(SqlTransform.query(options.getQuery().get()))

query 方法将 String 作为参数，这就是为什么您需要获取 ValueProvider 选项的 String 值。