?????? ??????????

[??????]

????????????????????????????????????????????????????????????????????????????????????????
????????????????????????????????


??????


 




------------------ ???????? ------------------
??????:                                                                         
                                               "user-zh"                        
                                                            
<tsreape...@gmail.com&gt;;
????????:&nbsp;2022??1??11??(??????) ????11:19
??????:&nbsp;"flink??????????"<user-zh@flink.apache.org&gt;;

????:&nbsp;Re: ??????????



Hi??

env.setRuntimeMode(RuntimeExecutionMode.BATCH); ?????????? table environment
???????????????????? table environment ???? streaming ?????????????????? Flink 
&gt;= 1.14??



?????? <2572805...@qq.com.invalid&gt; ??2022??1??10?????? 15:03??????

&gt; ????????????????:&nbsp; 
??????fromElements??????????????????????????????????????????????????
&gt; ??????????????????????????????????????????????
&gt; ??????????
&gt; StreamExecutionEnvironment env =
&gt; StreamExecutionEnvironment.getExecutionEnvironment();
&gt; StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
&gt; env.setRuntimeMode(RuntimeExecutionMode.BATCH);&nbsp; //????????????????
&gt;
&gt; DataStreamSource<Tuple2<String, Integer&amp;gt;&amp;gt; source =
&gt; env.fromElements(Tuple2.of("aa", 1),
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; Tuple2.of("aa", 
2),Tuple2.of("aa", 3),Tuple2.of("bb",
&gt; 2),Tuple2.of("bb", 3),Tuple2.of("bb", 4));
&gt; Table table = tenv.fromDataStream(source,
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; Schema.newBuilder()
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 .column("f0", "STRING")
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 .column("f1", "INTEGER")
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 .build());
&gt; tenv.createTemporaryView("test",table);
&gt; //????????sql????
&gt; tenv.createTemporarySystemFunction("Average", avg5.Average.class);
&gt; tenv.executeSql("SELECT f0,Average(f1) as rbm FROM test group by
&gt; f0").print();
&gt;
&gt; ??????
&gt; +----+--------------------------------+--------------------------------+
&gt; | op 
|&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 f0 
|&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 rbm |
&gt; +----+--------------------------------+--------------------------------+
&gt; | +I 
|&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 aa | [0, 0, 0, 0, 1, 0, 0, 0, 0,... |
&gt; | -U 
|&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 aa | [0, 0, 0, 0, 1, 0, 0, 0, 0,... |
&gt; | +U 
|&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 aa | [0, 0, 0, 0, 1, 0, 0, 0, 0,... |
&gt; | -U 
|&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 aa | [0, 0, 0, 0, 1, 0, 0, 0, 0,... |
&gt; | +U 
|&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 aa | [0, 0, 0, 0, 1, 0, 0, 0, 0,... |
&gt; | +I 
|&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 bb | [0, 0, 0, 0, 1, 0, 0, 0, 0,... |
&gt; | -U 
|&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 bb | [0, 0, 0, 0, 1, 0, 0, 0, 0,... |
&gt; | +U 
|&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 bb | [0, 0, 0, 0, 1, 0, 0, 0, 0,... |
&gt; | -U 
|&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 bb | [0, 0, 0, 0, 1, 0, 0, 0, 0,... |
&gt; | +U 
|&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 bb | [0, 0, 0, 0, 1, 0, 0, 0, 0,... |
&gt; +----+--------------------------------+--------------------------------+
&gt; ??????????????????????????????????????????????????????
&gt;
&gt; 
??????????????batch??????????????????????????????????????????????????????????????????
&gt;
&gt; 
????????????????????????????????????????????????????????????????????????????????????????????????
&gt;
&gt; ??????
&gt;
&gt;
&gt; &amp;nbsp;