Hi all, 业务上一般是可以避免动态类型的UDF的,如果有刚需,1.11已经支持了[1],文档还在路上,一个简单的例子根据第一个参数来推断返回类型:
@Override public TypeInference getTypeInference(DataTypeFactory typeFactory) { return TypeInference.newBuilder() .outputTypeStrategy(TypeStrategies.argument(0)) .build(); } [1]https://issues.apache.org/jira/browse/FLINK-15487 Best, Jingsong Lee On Tue, Jun 9, 2020 at 4:57 PM kcz <573693...@qq.com> wrote: > 有道理呀,我直接map就好 。tks。 > > > > > ------------------ 原始邮件 ------------------ > 发件人: "1048262223"<1048262...@qq.com>; > 发送时间: 2020年6月9日(星期二) 下午4:51 > 收件人: "user-zh"<user-zh@flink.apache.org>; > > 主题: 回复: Flink SQL UDF 动态类型 > > > > Hi > > > 这个可以通过返回一个通用类型比如Map来实现。 > > > Best, > Yichao Yang > > > > > ------------------&nbsp;原始邮件&nbsp;------------------ > 发件人:&nbsp;"kcz"<573693...@qq.com&gt;; > 发送时间:&nbsp;2020年6月9日(星期二) 下午4:49 > 收件人:&nbsp;"user-zh"<user-zh@flink.apache.org&gt;; > > 主题:&nbsp;回复: Flink SQL UDF 动态类型 > > > > 动态类型这个我们其实也挺需要的,比如我写一个列转行的udf,这个时候我需要确认返回的字段个数以及返回的类型,如果增加字段等,就需要更多udf来实现。 > > > > > ------------------&amp;nbsp;原始邮件&amp;nbsp;------------------ > 发件人:&amp;nbsp;"Benchao Li"<libenc...@apache.org&amp;gt;; > 发送时间:&amp;nbsp;2020年6月9日(星期二) 下午2:47 > 收件人:&amp;nbsp;"user-zh"<user-zh@flink.apache.org&amp;gt;; > > 主题:&amp;nbsp;Re: Flink SQL UDF 动态类型 > > > > 我也感觉这种场景可以用一个pb format来搞比较好一些,其实我们内部也是这么用的。 > 我们的format实现也是拿到pb编译后的class,根据这个class推导出来table的schema。 > 这个过程主要是需要定义清楚pb的所有类型到flink类型的映射关系。 > > 之前也考虑过在社区讨论下是否有必要支持一下pb format。如果你们都有这个诉求, > 我们可以先建一个jira,然后再jira里面讨论具体的需求和实现。 > > 1048262223 <1048262...@qq.com&amp;gt; 于2020年6月9日周二 下午2:23写道: > > &amp;gt; Hi > &amp;gt; > &amp;gt; > &amp;gt; > &amp;gt; > 我们是用到了pb作为源数据的,我们的方法是在程序初始化的时候根据pb的schema(descriptor)解析出TypeInformation,然后再env.addSource().returns()内指定我们解析出的TypeInformation,这个TypeInformation可以是任何动态类型的。 > &amp;gt; > &amp;gt; > &amp;gt; > &amp;gt; > 但是你的场景是使用udf,根据你发的udf示例来看,我明白你想要动态类型输出,但是实际上我不了解你们的场景,不明白什么样的场景需要用到这种处理,以及会用到一个udf产出不同的结果,因为我理解为了udf的管理方便、可读性以及可维护性,udf的输出参数类型应该定下来比较好一点。 > &amp;gt; > &amp;gt; > &amp;gt; 如果有理解不对之处,敬请指出。 > &amp;gt; > &amp;gt; > &amp;gt; Best, > &amp;gt; Yichao Yang > &amp;gt; > &amp;gt; > &amp;gt; > &amp;gt; > &amp;gt; > ------------------&amp;amp;nbsp;原始邮件&amp;amp;nbsp;------------------ > &amp;gt; 发件人:&amp;amp;nbsp;"forideal"<fszw...@163.com > &amp;amp;gt;; > &amp;gt; 发送时间:&amp;amp;nbsp;2020年6月9日(星期二) 中午1:33 > &amp;gt; 收件人:&amp;amp;nbsp;"user-zh"<user-zh@flink.apache.org > &amp;amp;gt;; > &amp;gt; > &amp;gt; 主题:&amp;amp;nbsp;Flink SQL UDF 动态类型 > &amp;gt; > &amp;gt; > &amp;gt; > &amp;gt; 你好,我的朋友: > &amp;gt; > &amp;gt; > &amp;gt; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; > 我使用的是 Flink 1.10 Blink Planer。 > &amp;gt; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; > 我想构造一个Flink UDF ,这个 UDF 可以根据不同的参数返回不同的类型。 > &amp;gt; > &amp;gt; > &amp;gt; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; > 为什么我想要这个功能: > &amp;gt; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; > 场景1: 我的数据是一个 pb 的 bytes,我想从里面获取数据,如果统一的返回 > &amp;gt; string,后面还需要 cast 比较繁琐,如果使用 get_int、get_double、get_string > 这样的方式,实现起来又非常多 > &amp;gt; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; > 场景2: 我的数据是一个 Json ,问题同上。 > &amp;gt; &amp;amp;nbsp; > &amp;gt; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; > 在场景1中,我改了下 Flink 的源码,在 ScalarFunction > &amp;gt; 中加了一个初始化方法,在Flink 初始化 scalar function 的时候,进行相关的初始化 > &amp;gt; @Override > &amp;gt; public void initialize(LogicalType[] sqlTypes, String[] > paramNames) { > &amp;gt; // 在这个函数里面做一些事情,比如,我可以根据 paramNames 去取 pb 的 schema > 信息,拿到类型信息,这样就可以动态的设置类型 > &amp;gt; } > &amp;gt; &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; > 这个方法很有效果,他帮我们 workaround 了一段时间,目前依然work。只是有些不是那么优雅。 > &amp;gt; &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; 这个case > 就是我想要的一个,不过,目前这个会返回 RAW('java.lang.Object', ?) > &amp;gt; 这个类型不进行 cast 是无法直接使用的。 > &amp;gt; public class TimestampTest extends ScalarFunction { > &amp;gt; > &amp;gt; public Object eval(long timestamp, String pattern, int num) { > &amp;gt; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; > Timestamp timestamp1 = new > &amp;gt; Timestamp(timestamp); > &amp;gt; SimpleDateFormat sdf = new SimpleDateFormat(pattern); > &amp;gt; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; > if (num < 4) { > &amp;gt; //返回 STRING 类型 > &amp;gt; return String.valueOf(timestamp); > &amp;gt; } > &amp;gt; if (num < 6) { > &amp;gt; //返回 BIGINT > &amp;gt; return timestamp - 100; > &amp;gt; } > &amp;gt; if (num < 8) { > &amp;gt; //返回 DOUBLE > &amp;gt; double ss = 0.9; > &amp;gt; > &amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp;&amp;amp;nbsp; > return > &amp;gt; (double) timestamp + ss; > &amp;gt; } > &amp;gt; //返回 STRING > &amp;gt; return sdf.format(timestamp1); > &amp;gt; } > &amp;gt; } -- Best, Jingsong Lee