如何在Apache Beam的SqlTransform中传递运行时查询？

问题背景 #

我希望运行Dataflow作业时动态传递SQL查询语句，目前硬编码查询到SqlTransform中可以正常工作，但业务场景需要在运行时传递查询，尝试用ValueProvider作为输入时出现编译错误。

硬编码查询的可行示例 #

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

使用ValueProvider时的编译错误 #

尝试通过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>)

解决方案 #

Apache Beam的SqlTransform.query()方法仅接受String类型参数，不直接支持ValueProvider——原因是SQL查询需要在作业构建阶段（编译时）完成解析、验证和执行计划生成，而ValueProvider的值要到运行时才确定，两者存在阶段冲突。根据不同场景，可采用以下方案：

1. 作业启动前注入查询（适用于非模板/模板启动传参场景）

如果查询在作业提交或启动时就能确定（比如通过命令行参数传递），可在Pipeline构建阶段将ValueProvider转换为String后传入SqlTransform。注意：此方法仅能在作业提交的本地执行阶段调用ValueProvider.get()，分布式运行时不可调用。

示例代码：

public interface MyOptions extends PipelineOptions {
  ValueProvider<String> getQuery();
  void setQuery(ValueProvider<String> query);
}

public static void main(String[] args) {
  MyOptions options = PipelineOptionsFactory.fromArgs(args).withValidation().as(MyOptions.class);
  Pipeline pipeline = Pipeline.create(options);
  
  // 仅在作业提交阶段（本地运行）调用get()获取查询
  String query = options.getQuery().get();
  PCollection<Row> rows = ...; // 输入数据集
  PCollection<Row> result = rows.apply(SqlTransform.query(query));
  
  pipeline.run();
}

2. 自定义DoFn实现运行时SQL解析（适用于动态查询场景）

如果查询必须在运行时动态获取（比如来自外部存储或数据流），则需要放弃SqlTransform，自行集成SQL解析引擎（如Apache Calcite），在DoFn中完成查询解析和数据处理。这种方式需要自行实现聚合、分组等逻辑，复杂度较高。

示例思路：

rows.apply(ParDo.of(new DoFn<Row, Row>() {
  private transient CalciteConnection connection;
  
  @Setup
  public void setup() throws SQLException {
    // 初始化Calcite连接用于SQL解析
    Properties props = new Properties();
    props.setProperty("model", "inline:{ ... }"); // 定义数据模型
    connection = DriverManager.getConnection("jdbc:calcite:", props);
  }
  
  @ProcessElement
  public void processElement(ProcessContext ctx) throws SQLException {
    MyOptions options = ctx.getPipelineOptions().as(MyOptions.class);
    String query = options.getQuery().get();
    
    // 解析并执行查询，处理数据
    try (Statement stmt = connection.createStatement()) {
      ResultSet rs = stmt.executeQuery(query);
      // 将ResultSet转换为Row输出
    }
  }
}));

3. Dataflow模板参数化查询

如果使用Dataflow模板，可将查询作为模板参数，在模板启动时传入，再转换为String传入SqlTransform。模板启动时的参数会在作业启动阶段确定，符合SqlTransform对编译时解析的要求。

内容的提问来源于stack exchange，提问作者NIKHIL SUTHAR