Flink DataStream转Table 使用SQL：select /where/ group by/toAppendStream/toRetractStream

妖狐艹你老母 2021-09-25 02:04 801阅读 0赞

### 方法1 registerDataStream  ###

package SQL;
    
    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.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import java.util.Date;
    import java.util.Random;
    import java.util.concurrent.TimeUnit;
    
    /**
     * @Author you guess
     * @Date 2020/6/15 12:08
     * @Version 1.0
     * @Desc
     */
    
    public class DataStreamSql1 {
        private static final Logger LOG = LoggerFactory.getLogger(MinMinByMaxMaxBy.MinMinByMaxMaxByTest.class);
        private static final String[] TYPE = {"a苹果", "b梨", "c西瓜", "d葡萄", "e火龙果"};
    
        public static void main(String[] args) throws Exception {
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            //TableEnvironment tEnv = TableEnvironment.create(TableConfig.getDefault());//报错
            //TableEnvironment tEnv = TableEnvironment.create(env);//报错
            EnvironmentSettings envSettings = EnvironmentSettings.newInstance().useOldPlanner().inStreamingMode().build();
            //TableEnvironment tEnv = TableEnvironment.create(envSettings);//TableEnvironment没有fromDataStream方法;没有registerDataStream方法；有tEnv.registerTable();tEnv.registerTableSink();tEnv.registerTableSource();
            StreamTableEnvironment stEnv = StreamTableEnvironment.create(env, envSettings);
    
            //添加自定义数据源,每秒发出一笔订单信息{商品名称,商品数量}
            DataStreamSource<Order> orderSource = 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() + ",提交元素:" + order);
                        ctx.collect(order);
                    }
                }
    
                @Override
                public void cancel() {
                    isRunning = false;
                }
    
            }, "order-info");
    
    
            //tableA是表名
            stEnv.registerDataStream("tableA", orderSource, "name,qtty");
            //这里查询的表名需要用前面注册好的表名，不然报错
            Table result = stEnv.sqlQuery("SELECT name,qtty FROM tableA");
            //Table result = stEnv.sqlQuery("SELECT * FROM tableA1");//Caused by: org.apache.calcite.sql.validate.SqlValidatorException: Object 'tableA1' not found
    
            stEnv.toAppendStream(result, Order.class).print();
    
            // after the table program is converted to DataStream program,
            // we must use `env.execute()` to submit the job.
            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 +
                        '}';
            }
        }
    }

输出：

Wed Jan 06 14:46:45 CST 2021,提交元素:Order\{name='c西瓜', qtty=2\}  
7> Order\{name='c西瓜', qtty=2\}  
Wed Jan 06 14:46:46 CST 2021,提交元素:Order\{name='a苹果', qtty=3\}  
8> Order\{name='a苹果', qtty=3\}  
Wed Jan 06 14:46:47 CST 2021,提交元素:Order\{name='d葡萄', qtty=3\}  
9> Order\{name='d葡萄', qtty=3\}  
Wed Jan 06 14:46:48 CST 2021,提交元素:Order\{name='e火龙果', qtty=6\}  
10> Order\{name='e火龙果', qtty=6\}

### 方法2使用fromDataStream ###

//DataStream转换成Table；fromDataStream有返回值；
            Table tableA = stEnv.fromDataStream(orderSource, "name,qtty");
    
            //这里的查询的表名需要是
            Table result = stEnv.sqlQuery("SELECT * FROM " + tableA);

输出同上。

### 注册或转换时使用其他字段名 ###

fromDataStream时，fields使用与类属性不同的字段名，会报错：

Exception in thread "main" org.apache.flink.table.api.ValidationException: name1 is not a field of type PojoType<SQL.DataStreamSql1$Order, fields = \[name: String, qtty: Long\]>. Expected: name, qtty\}

Table tableA = stEnv.fromDataStream(orderSource, "name1,qtty1");//Exception in thread "main" org.apache.flink.table.api.ValidationException: name1 is not a field of type PojoType<SQL.DataStreamSql1$Order, fields = [name: String, qtty: Long]>. Expected: name, qtty}
    
            //这里的查询的表名需要是
            Table result = stEnv.sqlQuery("SELECT * FROM " + tableA);

registerDataStream时，fields使用与类属性不同的字段名，会报错：

//tableA是表名
            stEnv.registerDataStream("tableA", orderSource, "name1,qtty1");
            //这里查询的表名需要用前面注册好的表名，不然报错
            Table result = stEnv.sqlQuery("SELECT name1,qtty1 FROM tableA");

报错Exception in thread "main" org.apache.flink.table.api.TableException: POJO does not define field name: name1

//tableA是表名
            stEnv.registerDataStream("tableA", orderSource, "name as name1,qtty as qtty1");
            //这里查询的表名需要用前面注册好的表名，不然报错
            Table result = stEnv.sqlQuery("SELECT name1,qtty1 FROM tableA");

### 注册或转换时不使用字段名 ###

Table tableA = stEnv.fromDataStream(orderSource);
            Table result = stEnv.sqlQuery("SELECT name,qtty FROM " + tableA + " where qtty > 5");
            //Table result = stEnv.sqlQuery("SELECT f0,f1 FROM " + tableA + " where f1 > 5");//报错Exception in thread "main" org.apache.flink.table.api.ValidationException: SQL validation failed. From line 1, column 40 to line 1, column 41: Column 'f1' not found in any table

或者

//tableA是表名
            stEnv.registerDataStream("tableA", orderSource);
            //这里查询的表名需要用前面注册好的表名，不然报错
            Table result = stEnv.sqlQuery("SELECT name,qtty FROM tableA where qtty > 5");

以上均是输出：

Wed Jan 06 15:07:32 CST 2021,提交元素:Order\{name='c西瓜', qtty=0\}  
Wed Jan 06 15:07:33 CST 2021,提交元素:Order\{name='a苹果', qtty=5\}  
Wed Jan 06 15:07:34 CST 2021,提交元素:Order\{name='b梨', qtty=0\}  
Wed Jan 06 15:07:35 CST 2021,提交元素:Order\{name='c西瓜', qtty=9\}  
9> Order\{name='c西瓜', qtty=9\}  
Wed Jan 06 15:07:36 CST 2021,提交元素:Order\{name='a苹果', qtty=8\}  
10> Order\{name='a苹果', qtty=8\}  
Wed Jan 06 15:07:37 CST 2021,提交元素:Order\{name='d葡萄', qtty=3\}  
Wed Jan 06 15:07:38 CST 2021,提交元素:Order\{name='e火龙果', qtty=0\}  
Wed Jan 06 15:07:39 CST 2021,提交元素:Order\{name='d葡萄', qtty=4\}  
Wed Jan 06 15:07:40 CST 2021,提交元素:Order\{name='e火龙果', qtty=1\}  
Wed Jan 06 15:07:41 CST 2021,提交元素:Order\{name='e火龙果', qtty=6\}  
3> Order\{name='e火龙果', qtty=6\}

//这里查询的表名需要用前面注册好的表名，不然报错
            Table result = stEnv.sqlQuery("SELECT name,qtty FROM tableA where qtty > 5");
            stEnv.toAppendStream(result, Order.class).print();
    //        Thu Jan 07 17:13:08 CST 2021,提交元素:Order{name='d葡萄', qtty=2}
    //        Thu Jan 07 17:13:09 CST 2021,提交元素:Order{name='d葡萄', qtty=9}
    //        6> Order{name='d葡萄', qtty=9}
    //        Thu Jan 07 17:13:10 CST 2021,提交元素:Order{name='b梨', qtty=9}
    //        7> Order{name='b梨', qtty=9}
    //        Thu Jan 07 17:13:11 CST 2021,提交元素:Order{name='b梨', qtty=8}
    //        8> Order{name='b梨', qtty=8}
    //        Thu Jan 07 17:13:12 CST 2021,提交元素:Order{name='c西瓜', qtty=0}
    //        Thu Jan 07 17:13:13 CST 2021,提交元素:Order{name='c西瓜', qtty=0}
    //        Thu Jan 07 17:13:14 CST 2021,提交元素:Order{name='d葡萄', qtty=1}
    //        Thu Jan 07 17:13:15 CST 2021,提交元素:Order{name='b梨', qtty=6}
    //        12> Order{name='b梨', qtty=6}
    
    
            //stEnv.toRetractStream(result, Order.class).print();
    //        Thu Jan 07 17:12:12 CST 2021,提交元素:Order{name='c西瓜', qtty=9}
    //        5> (true,Order{name='c西瓜', qtty=9})//true是插入，false是撤回
    //        Thu Jan 07 17:12:13 CST 2021,提交元素:Order{name='e火龙果', qtty=9}
    //        6> (true,Order{name='e火龙果', qtty=9})
    //        Thu Jan 07 17:12:14 CST 2021,提交元素:Order{name='c西瓜', qtty=0}
    //        Thu Jan 07 17:12:15 CST 2021,提交元素:Order{name='e火龙果', qtty=1}
    //        Thu Jan 07 17:12:16 CST 2021,提交元素:Order{name='e火龙果', qtty=6}
    //        9> (true,Order{name='e火龙果', qtty=6})
    //        Thu Jan 07 17:12:17 CST 2021,提交元素:Order{name='c西瓜', qtty=3}
    //        Thu Jan 07 17:12:18 CST 2021,提交元素:Order{name='e火龙果', qtty=5}
    //        Thu Jan 07 17:12:19 CST 2021,提交元素:Order{name='b梨', qtty=6}
    //        12> (true,Order{name='b梨', qtty=6})

### 使用group by  ###

Table result = stEnv.sqlQuery("SELECT name,sum(qtty) FROM tableA group by name");
            stEnv.toRetractStream(result, Row.class).print();//这里要用Row类型
            //stEnv.toAppendStream(result, Row.class).print();//报错Exception in thread "main" org.apache.flink.table.api.ValidationException: Table is not an append-only table. Use the toRetractStream() in order to handle add and retract messages.

Wed Jan 06 15:16:30 CST 2021,提交元素:Order\{name='b梨', qtty=7\}  
5> (true,b梨,7)  
Wed Jan 06 15:16:31 CST 2021,提交元素:Order\{name='c西瓜', qtty=2\}  
8> (true,c西瓜,2)  
Wed Jan 06 15:16:32 CST 2021,提交元素:Order\{name='d葡萄', qtty=7\}  
3> (true,d葡萄,7)  
Wed Jan 06 15:16:33 CST 2021,提交元素:Order\{name='c西瓜', qtty=1\}  
8> (false,c西瓜,2)  
8> (true,c西瓜,3)  
Wed Jan 06 15:16:34 CST 2021,提交元素:Order\{name='a苹果', qtty=2\}  
9> (true,a苹果,2)  
Wed Jan 06 15:16:35 CST 2021,提交元素:Order\{name='d葡萄', qtty=2\}  
3> (false,d葡萄,7)**//上次的值**  
3> (true,d葡萄,9**)//加上本次后的值**  
Wed Jan 06 15:16:36 CST 2021,提交元素:Order\{name='b梨', qtty=1\}  
5> (false,b梨,7)  
5> (true,b梨,8)  
Wed Jan 06 15:16:37 CST 2021,提交元素:Order\{name='b梨', qtty=2\}  
5> (false,b梨,8)  
5> (true,b梨,10)  
Wed Jan 06 15:16:38 CST 2021,提交元素:Order\{name='d葡萄', qtty=7\}  
3> (false,d葡萄,9)//**上次的值**  
3> (true,d葡萄,16)**)//加上本次后的值**  
Wed Jan 06 15:16:39 CST 2021,提交元素:Order\{name='e火龙果', qtty=9\}  
4> (true,e火龙果,9)  
Wed Jan 06 15:16:40 CST 2021,提交元素:Order\{name='c西瓜', qtty=1\}  
8> (false,c西瓜,3)  
8> (true,c西瓜,4)  
Wed Jan 06 15:16:41 CST 2021,提交元素:Order\{name='c西瓜', qtty=5\}  
8> (false,c西瓜,4)  
8> (true,c西瓜,9)  
Wed Jan 06 15:16:42 CST 2021,提交元素:Order\{name='c西瓜', qtty=9\}  
8> (false,c西瓜,9)  
8> (true,c西瓜,18)  
Wed Jan 06 15:16:43 CST 2021,提交元素:Order\{name='d葡萄', qtty=7\}  
3> (false,d葡萄,16)  
3> (true,d葡萄,23)

### 表转流toAppendStream/toRetractStream ###

toAppendStream适用于只是添加，无count()/group by 操作

toRetractStream适用于添加和撤回，有count()/group by 操作；第一个参数flag是true表示add，false表示retract

StreamTableEnvironment.java

/**
    	 * Converts the given {@link Table} into an append {@link DataStream} of a specified type.
    	 *
    	 * The {@link Table} must only have insert (append) changes. If the {@link Table} is also modified
    	 * by update or delete changes, the conversion will fail.
    	 *
    	 * <p>The fields of the {@link Table} are mapped to {@link DataStream} fields as follows:
    	 * <ul>
    	 *     <li>{@link org.apache.flink.types.Row} and {@link org.apache.flink.api.java.tuple.Tuple}
    	 *     types: Fields are mapped by position, field types must match.</li>
    	 *     <li>POJO {@link DataStream} types: Fields are mapped by field name, field types must match.</li>
    	 * </ul>
    	 *
    	 * @param table The {@link Table} to convert.
    	 * @param typeInfo The {@link TypeInformation} that specifies the type of the {@link DataStream}.
    	 * @param queryConfig The configuration of the query to generate.
    	 * @param <T> The type of the resulting {@link DataStream}.
    	 * @return The converted {@link DataStream}.
    	 */
    	<T> DataStream<T> toAppendStream(Table table, TypeInformation<T> typeInfo, StreamQueryConfig queryConfig);
    
    	/**
    	 * Converts the given {@link Table} into a {@link DataStream} of add and retract messages.
    	 * The message will be encoded as {@link Tuple2}. The first field is a {@link Boolean} flag,
    	 * the second field holds the record of the specified type {@link T}.
    	 *
    	 * A true {@link Boolean} flag indicates an add message, a false flag indicates a retract message.
    	 *
    	 * <p>The fields of the {@link Table} are mapped to {@link DataStream} fields as follows:
    	 * <ul>
    	 *     <li>{@link org.apache.flink.types.Row} and {@link org.apache.flink.api.java.tuple.Tuple}
    	 *     types: Fields are mapped by position, field types must match.</li>
    	 *     <li>POJO {@link DataStream} types: Fields are mapped by field name, field types must match.</li>
    	 * </ul>
    	 *
    	 * @param table The {@link Table} to convert.
    	 * @param clazz The class of the requested record type.
    	 * @param <T> The type of the requested record type.
    	 * @return The converted {@link DataStream}.
    	 */
    	<T> DataStream<Tuple2<Boolean, T>> toRetractStream(Table table, Class<T> clazz);

不同flink版本的语法差别挺大:

<flink.version>1.9.2</flink.version>

jdk1.8