Re: Flink SQL UDF 动态类型

Jingsong Li Tue, 09 Jun 2020 02:22:09 -0700

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。
>
>
>
>
> ------------------&nbsp;原始邮件&nbsp;------------------
> 发件人:&nbsp;"1048262223"<1048262...@qq.com&gt;;
> 发送时间:&nbsp;2020年6月9日(星期二) 下午4:51
> 收件人:&nbsp;"user-zh"<user-zh@flink.apache.org&gt;;
>
> 主题:&nbsp;回复： Flink SQL UDF 动态类型
>
>
>
> Hi
>
>
> 这个可以通过返回一个通用类型比如Map来实现。
>
>
> Best,
> Yichao Yang
>
>
>
>
> ------------------&amp;nbsp;原始邮件&amp;nbsp;------------------
> 发件人:&amp;nbsp;"kcz"<573693...@qq.com&amp;gt;;
> 发送时间:&amp;nbsp;2020年6月9日(星期二) 下午4:49
> 收件人:&amp;nbsp;"user-zh"<user-zh@flink.apache.org&amp;gt;;
>
> 主题:&amp;nbsp;回复： Flink SQL UDF 动态类型
>
>
>
> 动态类型这个我们其实也挺需要的，比如我写一个列转行的udf，这个时候我需要确认返回的字段个数以及返回的类型，如果增加字段等，就需要更多udf来实现。
>
>
>
>
> ------------------&amp;amp;nbsp;原始邮件&amp;amp;nbsp;------------------
> 发件人:&amp;amp;nbsp;"Benchao Li"<libenc...@apache.org&amp;amp;gt;;
> 发送时间:&amp;amp;nbsp;2020年6月9日(星期二) 下午2:47
> 收件人:&amp;amp;nbsp;"user-zh"<user-zh@flink.apache.org&amp;amp;gt;;
>
> 主题:&amp;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;amp;gt; 于2020年6月9日周二 下午2:23写道：
>
> &amp;amp;gt; Hi
> &amp;amp;gt;
> &amp;amp;gt;
> &amp;amp;gt;
> &amp;amp;gt;
> 我们是用到了pb作为源数据的，我们的方法是在程序初始化的时候根据pb的schema(descriptor)解析出TypeInformation，然后再env.addSource().returns()内指定我们解析出的TypeInformation，这个TypeInformation可以是任何动态类型的。
> &amp;amp;gt;
> &amp;amp;gt;
> &amp;amp;gt;
> &amp;amp;gt;
> 但是你的场景是使用udf，根据你发的udf示例来看，我明白你想要动态类型输出，但是实际上我不了解你们的场景，不明白什么样的场景需要用到这种处理，以及会用到一个udf产出不同的结果，因为我理解为了udf的管理方便、可读性以及可维护性，udf的输出参数类型应该定下来比较好一点。
> &amp;amp;gt;
> &amp;amp;gt;
> &amp;amp;gt; 如果有理解不对之处，敬请指出。
> &amp;amp;gt;
> &amp;amp;gt;
> &amp;amp;gt; Best,
> &amp;amp;gt; Yichao Yang
> &amp;amp;gt;
> &amp;amp;gt;
> &amp;amp;gt;
> &amp;amp;gt;
> &amp;amp;gt;
> ------------------&amp;amp;amp;nbsp;原始邮件&amp;amp;amp;nbsp;------------------
> &amp;amp;gt; 发件人:&amp;amp;amp;nbsp;"forideal"<fszw...@163.com
> &amp;amp;amp;gt;;
> &amp;amp;gt; 发送时间:&amp;amp;amp;nbsp;2020年6月9日(星期二) 中午1:33
> &amp;amp;gt; 收件人:&amp;amp;amp;nbsp;"user-zh"<user-zh@flink.apache.org
> &amp;amp;amp;gt;;
> &amp;amp;gt;
> &amp;amp;gt; 主题:&amp;amp;amp;nbsp;Flink SQL UDF 动态类型
> &amp;amp;gt;
> &amp;amp;gt;
> &amp;amp;gt;
> &amp;amp;gt; 你好，我的朋友：
> &amp;amp;gt;
> &amp;amp;gt;
> &amp;amp;gt;
> &amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;
> 我使用的是 Flink 1.10 Blink Planer。
> &amp;amp;gt;
> &amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;
> 我想构造一个Flink UDF ，这个 UDF 可以根据不同的参数返回不同的类型。
> &amp;amp;gt;
> &amp;amp;gt;
> &amp;amp;gt;
> &amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;
> 为什么我想要这个功能：
> &amp;amp;gt;
> &amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;
> 场景1: 我的数据是一个 pb 的 bytes，我想从里面获取数据，如果统一的返回
> &amp;amp;gt; string，后面还需要 cast 比较繁琐，如果使用 get_int、get_double、get_string
> 这样的方式，实现起来又非常多
> &amp;amp;gt;
> &amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;
> 场景2: 我的数据是一个 Json ，问题同上。
> &amp;amp;gt; &amp;amp;amp;nbsp;
> &amp;amp;gt;
> &amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;
> 在场景1中，我改了下 Flink 的源码，在 ScalarFunction
> &amp;amp;gt; 中加了一个初始化方法，在Flink 初始化 scalar function 的时候，进行相关的初始化
> &amp;amp;gt; @Override
> &amp;amp;gt; public void initialize(LogicalType[] sqlTypes, String[]
> paramNames) {
> &amp;amp;gt; // 在这个函数里面做一些事情，比如，我可以根据 paramNames 去取 pb 的 schema
> 信息，拿到类型信息，这样就可以动态的设置类型
> &amp;amp;gt; }
> &amp;amp;gt; &amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;
> 这个方法很有效果，他帮我们 workaround 了一段时间，目前依然work。只是有些不是那么优雅。
> &amp;amp;gt; &amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp; 这个case
> 就是我想要的一个，不过，目前这个会返回 RAW('java.lang.Object', ?)
> &amp;amp;gt; 这个类型不进行 cast 是无法直接使用的。
> &amp;amp;gt; public class TimestampTest extends ScalarFunction {
> &amp;amp;gt;
> &amp;amp;gt; public Object eval(long timestamp, String pattern, int num) {
> &amp;amp;gt;
> &amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;
> Timestamp timestamp1 = new
> &amp;amp;gt; Timestamp(timestamp);
> &amp;amp;gt; SimpleDateFormat sdf = new SimpleDateFormat(pattern);
> &amp;amp;gt;
> &amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;
> if (num < 4) {
> &amp;amp;gt; //返回 STRING 类型
> &amp;amp;gt; return String.valueOf(timestamp);
> &amp;amp;gt; }
> &amp;amp;gt; if (num < 6) {
> &amp;amp;gt; //返回 BIGINT
> &amp;amp;gt; return timestamp - 100;
> &amp;amp;gt; }
> &amp;amp;gt; if (num < 8) {
> &amp;amp;gt; //返回 DOUBLE
> &amp;amp;gt; double ss = 0.9;
> &amp;amp;gt;
> &amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;&amp;amp;amp;nbsp;
> return
> &amp;amp;gt; (double) timestamp + ss;
> &amp;amp;gt; }
> &amp;amp;gt; //返回 STRING
> &amp;amp;gt; return sdf.format(timestamp1);
> &amp;amp;gt; }
> &amp;amp;gt; }



-- 
Best, Jingsong Lee

Re: Flink SQL UDF 动态类型

Reply via email to