Apache Beam SQL中row_number()函数报错，请求修复方案

这个问题我之前也碰到过，核心原因是Apache Beam SQL对ROW_NUMBER()这类窗口函数的支持存在版本兼容性或Runner适配的限制，下面给你几个可行的修复方案：

方案一：升级到支持窗口函数的Beam版本 #

Apache Beam对SQL窗口函数的支持是逐步完善的，2.10.0版本才开始初步支持，2.20.0+的稳定版本兼容性更好。如果你的Beam版本较低，建议直接升级项目依赖（比如Maven中的beam-sql、beam-runners-xxx等依赖），新版本对窗口函数的解析逻辑做了很多优化，大概率能解决这个翻译失败的问题。

方案二：绕过SQL，用Beam核心API实现行号逻辑 #

如果升级版本后还是不行，可以直接用Beam的Dataflow API来实现分组排序生成行号的逻辑，灵活性更高：

PCollection<Row> rwrtg = rrtg
    // 先按appId分组（假设appId是字符串类型，根据实际类型调整）
    .apply(WithKeys.of((Row row) -> row.getString("appId")))
    .apply(GroupByKey.create())
    .apply(ParDo.of(new DoFn<KV<String, Iterable<Row>>, Row>() {
        @ProcessElement
        public void processElement(ProcessContext c) {
            String appId = c.element().getKey();
            Iterable<Row> rows = c.element().getValue();
            
            // 将分组内的元素按rating升序排序
            List<Row> sortedRows = Streams.stream(rows)
                .sorted(Comparator.comparing(row -> row.getInt("rating")))
                .collect(Collectors.toList());
            
            // 为每个元素生成行号并构造新Row输出
            int rank = 1;
            for (Row row : sortedRows) {
                // 构建包含issue_rank字段的新Schema
                Schema newSchema = row.getSchema().addField(
                    Field.of("issue_rank", FieldType.INT32)
                );
                Row newRow = Row.withSchema(newSchema)
                    .addValues(row.getValues().toArray())
                    .addValue(rank++)
                    .build();
                c.output(newRow);
            }
        }
    }));

方案三：检查Runner的兼容性配置 #

如果你使用的是特定的执行Runner（比如Flink、Spark或Dataflow），需要确保Runner本身支持Beam SQL的窗口函数：

• Flink Runner：确认Flink版本与Beam版本兼容，并且提交作业时开启SQL相关的配置项；
• Dataflow Runner：确保使用的Dataflow SDK版本支持Beam SQL窗口函数，作业配置中无需额外关闭SQL特性；
• Spark Runner：检查Spark版本是否适配当前Beam版本，Spark对窗口函数的支持相对成熟，但需要Beam做好适配。

方案四：尝试显式指定窗口框架语法 #

有时候Beam SQL对窗口函数的语法解析比较严格，你可以尝试显式写出默认的窗口框架（虽然报错信息里已经包含，但试试显式声明）：

select appId, row_number() over (
    partition by appId 
    order by rating asc 
    rows between unbounded preceding and current row
) as issue_rank from trrtg

这个方案不一定能解决所有情况，但可以排除语法解析层面的问题。

内容的提问来源于stack exchange，提问作者Syed Mohammed Mehdi