?????? Flink SQL UDF ????????

[kcz]

????????????????map???? ??tks??




------------------ ???????? ------------------
??????:&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; }