Flink 使用TableFunction getResultType返回Row类型

爱被打了一巴掌 2021-09-25 07:02 322阅读 0赞

Flink1.9.3

package SQL_split_index;
    
    import org.apache.flink.api.common.typeinfo.TypeInformation;
    import org.apache.flink.api.common.typeinfo.Types;
    import org.apache.flink.streaming.api.datastream.DataStreamSource;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.source.SourceFunction;
    import org.apache.flink.table.api.EnvironmentSettings;
    import org.apache.flink.table.api.Table;
    import org.apache.flink.table.api.java.StreamTableEnvironment;
    import org.apache.flink.table.functions.TableFunction;
    import org.apache.flink.types.Row;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import java.util.Date;
    import java.util.Random;
    import java.util.concurrent.TimeUnit;
    
    /**
     * @Author you guess
     * @Date 2021/4/12 17:12
     * @Version 1.0
     * @Desc
     */
    public class DataStreamSql1_Split_Lateral_Table {
        private static final Logger LOG = LoggerFactory.getLogger(DataStreamSql1_Split_Lateral_Table.class);
        private static final String[] TYPE = {"a_苹果", "b_梨", "c_西瓜", "d_葡萄", "e_火龙果"};
    
        public static void main(String[] args) throws Exception {
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            EnvironmentSettings envSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
            StreamTableEnvironment stEnv = StreamTableEnvironment.create(env, envSettings);
    
            //添加自定义数据源,每秒发出一笔订单信息{商品名称,商品数量}
            DataStreamSource<Order> orderSourceA = env.addSource(new SourceFunction<Order>() {
                private volatile boolean isRunning = true;
                private final Random random = new Random();
    
                @Override
                public void run(SourceContext<Order> ctx) throws Exception {
                    while (isRunning) {
                        TimeUnit.SECONDS.sleep(1);
                        Order order = new Order(TYPE[random.nextInt(TYPE.length)], Long.valueOf(random.nextInt(10)));
                        System.out.println(new Date() + ",orderSourceA提交元素:" + order);
                        ctx.collect(order);
                    }
                }
    
                @Override
                public void cancel() {
                    isRunning = false;
                }
    
            }, "order-infoA");
    
    
            stEnv.registerDataStream("tableA", orderSourceA);
            stEnv.registerFunction("SplitTF1", new SplitTF1());
            stEnv.registerFunction("SplitTF2", new SplitTF2());
            stEnv.registerFunction("SplitTF3", new SplitTF3());
            stEnv.registerFunction("SplitTF4", new SplitTF4());
    
    //        Table result = stEnv.sqlQuery("SELECT A.name,A.qtty,split_index(name,'_',0) as name0,split_index(name,'_',1) as name1  from tableA A");
    //        Table result = stEnv.sqlQuery("SELECT name,qtty,B.name0,B.name1,B.name1Length  from tableA, lateral table (SplitTF1(name)) as B(name0,name1,name1Length)");//报错
    //        Table result = stEnv.sqlQuery("SELECT name,qtty,B.name0  from tableA, lateral table (SplitTF2(name)) as B(name0)");//正确
    //        Table result = stEnv.sqlQuery("SELECT name,qtty,B.name0  from tableA, lateral table (SplitTF3(name)) as B(name0)");//正确
    //        Table result = stEnv.sqlQuery("SELECT name,qtty,B.*  from tableA, lateral table (SplitTF4(name)) as B");//(true,e_火龙果,9,e,火龙果,3)//正确
    //        Table result = stEnv.sqlQuery("SELECT name,qtty,B.name1,B.name1Length from tableA, lateral table (SplitTF4(name)) as B(name0,name1,name1Length)");//正确
            Table result = stEnv.sqlQuery("SELECT name,qtty,B.name11,B.name11Length from tableA, lateral table (SplitTF4(name)) as B(name00,name11,name11Length)");//正确
            stEnv.toRetractStream(result, Row.class).print();//这里要用Row类型
            env.execute("Flink Streaming Java API Skeleton");
    
        }
    
    
        /**
         * Simple POJO.
         */
        public static class Order {
            public String name;
            public Long qtty;
    
    
            public Order() {
            }
    
            public Order(String name, Long qtty) {
                this.name = name;
                this.qtty = qtty;
            }
    
            @Override
            public String toString() {
                return "Order{" +
                        "name='" + name + '\'' +
                        ", qtty=" + qtty +
                        '}';
            }
        }
    
    
        /**
         * Simple POJO.
         */
        public static class Order2 {
            public String name0;
            public String name1;
            public Integer name1Length;
    
            public Order2(String name0, String name1, Integer name1Length) {
                this.name0 = name0;
                this.name1 = name1;
                this.name1Length = name1Length;
            }
    
    
            @Override
            public String toString() {
                return "Order2{" +
                        " name0='" + name0 + '\'' +
                        ", name1='" + name1 + '\'' +
                        ", name1Length='" + name1Length + '\'' +
                        '}';
            }
        }
    
    
        public static class Order3 {
            public String name0;
    
            public Order3(String name0) {
                this.name0 = name0;
            }
    
    
            @Override
            public String toString() {
                return "Order3{" +
                        " name0='" + name0 + '\'' +
                        '}';
            }
        }
    
        //报错
        public static class SplitTF1 extends TableFunction<Order2> {
            public void eval(String s) {
                collect(new Order2(s.split("_")[0], s.split("_")[1], s.split("_")[1].length()));
            }
        }
    
        //正确
        public static class SplitTF2 extends TableFunction<Order3> {
            public void eval(String s) {
                collect(new Order3(s.split("_")[0]));
            }
        }
    
    
        //正确
        public static class SplitTF3 extends TableFunction<Row> {
            public void eval(String s) {
                Row row = Row.of(s.split("_")[0]);
                collect(row);
            }
        }
    
    
        //正确
        public static class SplitTF4 extends TableFunction<Row> {
            public void eval(String s) {
                Row row = Row.of(s.split("_")[0], s.split("_")[1], s.split("_")[1].length());
                collect(row);
            }
    
            //这个方法很重要，不写的话会报错！！！！
            //SQL validation failed. From line 1, column 89 to line 1, column 111: List of column aliases must have same degree as table; table has 1 columns ('f0'), whereas alias list has 3 columns
            //at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.validate(FlinkPlannerImpl.scala:125)
            @Override
            public TypeInformation<Row> getResultType() {
                return Types.ROW(Types.STRING, Types.STRING, Types.INT);
            }
        }
    
    }

输出：

5> (true,c\_西瓜,7,西瓜,2)  
Tue Apr 13 11:32:41 CST 2021,orderSourceA提交元素:Order\{name='d\_葡萄', qtty=8\}  
6> (true,d\_葡萄,8,葡萄,2)  
Tue Apr 13 11:32:42 CST 2021,orderSourceA提交元素:Order\{name='c\_西瓜', qtty=3\}  
7> (true,c\_西瓜,3,西瓜,2)  
Tue Apr 13 11:32:43 CST 2021,orderSourceA提交元素:Order\{name='c\_西瓜', qtty=4\}  
8> (true,c\_西瓜,4,西瓜,2)  
Tue Apr 13 11:32:44 CST 2021,orderSourceA提交元素:Order\{name='e\_火龙果', qtty=5\}  
1> (true,e\_火龙果,5,火龙果,3)  
Tue Apr 13 11:32:45 CST 2021,orderSourceA提交元素:Order\{name='a\_苹果', qtty=7\}  
2> (true,a\_苹果,7,苹果,2)  
Tue Apr 13 11:32:46 CST 2021,orderSourceA提交元素:Order\{name='b\_梨', qtty=8\}  
3> (true,b\_梨,8,梨,1)  
Tue Apr 13 11:32:47 CST 2021,orderSourceA提交元素:Order\{name='e\_火龙果', qtty=8\}  
4> (true,e\_火龙果,8,火龙果,3)