Flink SQL UDF重复调用问题解决方案

原创

皮皮熊

发布于 2023-05-31 20:42:49

1.2K1

发布于 2023-05-31 20:42:49

文章被收录于专栏：大数据与实时计算大数据与实时计算

Flink SQL UDF重复调用/执行问题

UDF重复调用问题

UDF重复调用的问题在某些情况下可能会对Flink SQL用户造成困扰，例如下面的SQL语句：

SELECT my_map['key1'] as key1, my_map['key2'] as key2, my_map['key3'] as key3
FROM (
  SELECT dump_json_to_map(col1) as my_map
  FROM T
)

dump_json_to_map会被执行3次。分析对应的graph日志开看，Flink会把我们的代码反向'优化'成类似如下sql：

SELECT dump_json_to_map(col1)['key1'] as key1, dump_json_to_map(col1)['key2'] as key2, dump_json_to_map(col1)['key3'] as key3 FROM T

会造成性能和正确性的问题：

UDF包含计算密集型的逻辑，整个作业的性能就会受到很大影响
UDF是有状态的UDF(如链接Redis等外部存储)，则会导致重复计算，中间状态可能因为无法幂等的操作而被破坏，最终导致正确性出现问题flink 有状态udf 引起血案一

这个Flink社区已有对应的讨论，但是已知没有具体的后续，详见：FLINK-21573

解决方案

解决方案一

修改Flink内核源码，需要团队成员具备维护Flink内核的能力和权力。

参考这篇文章。

思路摘要：

复写udf的isDeterministic()方法
在CodeGeneratorContext中添加可重用的UDF表达式及其result term的容器
从ExprCodeGenerator入手(函数调用都属于RexCall)，找到UDF代码生成的方法，即BridgingFunctionGenUtil#generateScalarFunctionCall(),if (isDeterministic)块内的代码实现了UDF表达式重用，即重用生成的第一个result term。

解决方案二（推荐）

来自好友kyle大佬的实战经验：增加一层透传专用的UDTF。

实现参考：

public class PassThroughUdtf extends TableFunction<String> {
    private static final long serialVersionUID = 1093578798410129502L;

    // 仅为示例，需要根据自己的场景修改入参和输出的数据类型
    public void eval(String param){
        collect(param);
    }
}

然后改造下原有SQL

SELECT my_map['key1'] as key1, my_map['key2'] as key2, my_map['key3'] as key3
FROM (
  SELECT my_map 
  FROM T ,lateral table(passThrough(dump_json_to_map(col1))) as T(my_map)
)

增加PassThroughUdtf后对整体性能影响不大，就可以相对简单地解决UDF重复调用的问题。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

flink