devbox
zyabc
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Flink Table API & SQL 编程_flink1.16 table api demo

作者：zyabc | 2024-01-29 17:34:46

赞

踩

flink1.16 table api demo

目录

Maven依赖

DEMO1 消费Kafka主题数据，生产主题数据

DEMO2 Table与DataStream的互相转换

DEMO3 时间窗口

sql-client的使用

执行建表语句

filesystem数据源

kafka数据源

执行查询语句

Maven依赖

  1.     <dependencies>
  2.         <dependency>
  3.             <groupId>org.apache.flink</groupId>
  4.             <artifactId>flink-clients_2.11</artifactId>
  5.             <version>${flink.version}</version>
  6.         </dependency>
  7.         <dependency>
  8.             <groupId>org.apache.flink</groupId>
  9.             <artifactId>flink-table-api-java-bridge_2.11</artifactId>
  10.             <version>${flink.version}</version>
  11.         </dependency>
  12.         <dependency>
  13.             <groupId>org.apache.flink</groupId>
  14.             <artifactId>flink-table-planner-blink_2.11</artifactId>
  15.             <version>${flink.version}</version>
  16.         </dependency>
  17.         <dependency>
  18.             <groupId>org.apache.flink</groupId>
  19.             <artifactId>flink-streaming-scala_2.11</artifactId>
  20.             <version>${flink.version}</version>
  21.         </dependency>
  22.         <dependency>
  23.             <groupId>org.apache.flink</groupId>
  24.             <artifactId>flink-streaming-java_2.11</artifactId>
  25.             <version>${flink.version}</version>
  26.         </dependency>
  27.         <dependency>
  28.             <groupId>org.apache.flink</groupId>
  29.             <artifactId>flink-table-common</artifactId>
  30.             <version>${flink.version}</version>
  31.         </dependency>
  32.  
  33.         <dependency>
  34.             <groupId>org.apache.flink</groupId>
  35.             <artifactId>flink-connector-kafka_2.11</artifactId>
  36.             <version>${flink.version}</version>
  37.         </dependency>
  38.         <!--json-->
  39.         <dependency>
  40.             <groupId>org.apache.flink</groupId>
  41.             <artifactId>flink-json</artifactId>
  42.             <version>${flink.version}</version>
  43.         </dependency>
  44.         <!--csv-->
  45.         <dependency>
  46.             <groupId>org.apache.flink</groupId>
  47.             <artifactId>flink-csv</artifactId>
  48.             <version>${flink.version}</version>
  49.         </dependency>
  50.     </dependencies>

DEMO1 消费Kafka主题数据，生产主题数据

  1. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  2. import org.apache.flink.table.api.Table;
  3. import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
  4.  
  5. /**
  6.  * Description: 从Kafka主题kafkaInput读取数据，筛选后将指定数据写入kafkaOutput
  7.  * 数据样式：
  8.  * {"id":"1","user_id":"1001","status":"1"}
  9.  * {"id":"2","user_id":"1002","status":"1"}
  10.  * {"id":"3","user_id":"1003","status":"1"}
  11.  * {"id":"4","user_id":"1004","status":"1"}
  12.  * {"id":"5","user_id":"1005","status":"0"}
  13.  */
  14. public class Demo01 {
  15.     public static void main(String[] args) throws Exception {
  16.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  17.         StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
  18.         tableEnv.executeSql("CREATE TABLE kafka_input(id BIGINT,user_id BIGINT," +
  19.                 "status STRING) " +
  20.                 "WITH ('connector' = 'kafka','topic' = 'kafkaInput'," +
  21.                 "'properties.bootstrap.servers' = 'vm01:9092'," +
  22.                 "'properties.group.id' = 'testGroup'," +
  23.                 "'scan.startup.mode' = 'latest-offset'," +
  24.                 "'format' = 'json')");
  25.         Table table = tableEnv.sqlQuery("SELECT * FROM kafka_input WHERE status='1'");
  26.         tableEnv.executeSql("CREATE TABLE kafka_output(id BIGINT,user_id BIGINT," +
  27.                 "status STRING)" +
  28.                 " WITH ('connector'='kafka','topic'='kafkaOutput'," +
  29.                 "'properties.bootstrap.servers' = 'vm01:9092'," +
  30.                 "'format'='json'," +
  31.                 "'sink.partitioner'='round-robin')");
  32.         tableEnv.executeSql("INSERT INTO kafka_output SELECT * FROM " + table);
  33.         // 会报错：No operators defined in streaming topology. Cannot execute.但不影响程序运行
  34.         // tableEnv.execute("Run TableapiKafkaDemo");
  35.     }
  36. }

DEMO2 Table与DataStream的互相转换

  1. import org.apache.flink.streaming.api.datastream.DataStream;
  2. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  3. import org.apache.flink.table.api.Table;
  4. import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
  5.  
  6.  
  7. public class Demo02 {
  8.     public static void main(String[] args) throws Exception {
  9.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  10.         StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
  11.         // kafka table -> stream
  12. //        table2Stream(tEnv);
  13.         // stream -> table -> stream
  14.         stream2Table(env, tEnv);
  15.         env.execute();
  16.     }
  17.  
  18.     /**
  19.      * kafka table -> stream print()
  20.      * 数据格式：1,Jordan,male
  21.      * @param tEnv StreamTableEnvironment
  23.      */
  24.     private static void table2Stream(StreamTableEnvironment tEnv) {
  25.         String createTableSql = "CREATE TABLE user_info(" +
  26.                 "id INT," +
  27.                 "name STRING," +
  28.                 "gender STRING" +
  29.                 ")WITH(" +
  30.                 "'connector'='kafka'," +
  31.                 "'topic'='user_info'," +
  32.                 "'properties.bootstrap.servers'='vm01:9092'," +
  33.                 "'properties.group.id'='testGroup'," +
  34.                 "'scan.startup.mode'='latest-offset'," +
  35.                 "'format'='csv'" +
  36.                 ")";
  37.         tEnv.executeSql(createTableSql);
  38.         Table table = tEnv.sqlQuery("SELECT * FROM user_info");
  39.         tEnv.toDataStream(table).print();
  40.     }
  41.  
  42.     /**
  43.      * stream -> table -> stream
  44.      * 
  45.      * @param env  StreamExecutionEnvironment
  46.      * @param tEnv StreamTableEnvironment
  47.      */
  48.     private static void stream2Table(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
  49.         DataStream<String> dataStream = env.fromElements("Jordan", "Alice", "Tom");
  50.         Table inputTable = tEnv.fromDataStream(dataStream);
  51.         tEnv.createTemporaryView("INPUT_TABLE", inputTable);
  52.         Table resultTable = tEnv.sqlQuery("SELECT UPPER(f0) FROM INPUT_TABLE");
  53.         resultTable.printSchema();
  54.         tEnv.toDataStream(resultTable).print();
  55.     }
  56. }

DEMO3 时间窗口

建表时定义一个TIMESTAMP(3)数据格式的列作为时间窗口取值列，并通过：WATERMARK FOR row_time AS row_time - INTERVAL '5' SECOND 指定水位线为5秒，然后正常开窗和使用聚合函数。

  1. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  2. import org.apache.flink.table.api.Slide;
  3. import org.apache.flink.table.api.Table;
  4. import org.apache.flink.table.api.Tumble;
  5. import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
  6.  
  7. import static org.apache.flink.table.api.Expressions.$;
  8. import static org.apache.flink.table.api.Expressions.lit;
  9.  
  10. /**
  11.  * Kafka数据格式：1,jordan,79,2022-03-05 00:00:30
  12.  */
  13. public class Demo03 {
  14.     public static void main(String[] args) throws Exception {
  15.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  16.         StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
  17.         String createTableSql = "CREATE TABLE user_info(" +
  18.                 "id INT," +
  19.                 "name STRING," +
  20.                 "score INT," +
  21.                 "row_time TIMESTAMP(3)," +
  22.                 "WATERMARK FOR row_time AS row_time - INTERVAL '5' SECOND" +
  23.                 ")WITH(" +
  24.                 "'connector'='kafka'," +
  25.                 "'topic'='t_score'," +
  26.                 "'properties.bootstrap.servers'='vm01:9092'," +
  27.                 "'properties.group.id'='testGroup'," +
  28.                 "'scan.startup.mode'='latest-offset'," +
  29.                 "'format'='csv'" +
  30.                 ")";
  31.         tEnv.executeSql(createTableSql);
  32.         Table table = tEnv.sqlQuery("SELECT * FROM user_info");
  33.         Table resultTable = table.window(Tumble.over(lit(10).seconds()).on($("row_time")).as("w"))
  34. //                            table.window(Slide.over(lit(10).seconds()).every(lit(20).seconds()).on($("row_time")).as("w"))
  35.                 .groupBy($("name"), $("w"))
  36.                 .select($("name"), $("score").avg(), $("w").end().as("hour"));
  37.         resultTable.printSchema();
  38.         tEnv.toDataStream(resultTable).print();
  39.  
  40.         env.execute("Demo03");
  41.     }
  42. }

sql-client的使用

  1. # 启动集群
  2. ${FLINK_HOME}/bin/start-cluster.sh
  3. # 启动命令行
  4. ${FLINK_HOME}/bin/sql-client.sh

此时可以交互式执行sql语句。

此外，可以通过使用以下方式执行sql脚本

  1. # 其中：~/file.sql为可正常执行的sql脚本,-f也可换成--file
  2. ${FLINK_HOME}/bin/sql-client.sh -f ~/file.sql

执行建表语句

如果需要使用时间窗口，则必须指定watermark

filesystem数据源

  1. CREATE TABLE bid(
  2.     bidtime TIMESTAMP(3),
  3.     price DECIMAL(10, 2),
  4.     item STRING,
  5. 	WATERMARK FOR bidtime AS bidtime - INTERVAL '10' MINUTES
  6. ) WITH ( 
  7.     'connector' = 'filesystem',
  8.     'path' = '/root/data/bid.csv',
  9.     'format' = 'csv'
  10. );

kafka数据源

  1. CREATE TABLE user_info(
  2.     id INT,
  3.     name STRING,
  4.     score INT,
  5.     row_time TIMESTAMP(3),
  6.     WATERMARK FOR row_time AS row_time - INTERVAL '5' SECOND
  7. ) WITH (
  8.     'connector'='kafka',
  9.     'topic'='t_score',
  10.     'properties.bootstrap.servers'='vm01:9092',
  11.     'properties.group.id'='testGroup',
  12.     'scan.startup.mode'='latest-offset',
  13.     'format'='csv'
  14. );

执行查询语句

一般的查询语句与通用SQL类似，对于内置函数，可以通过SHOW FUNCTIONS进行查看。

下面演示的是窗口函数的使用。

  1. # 所查询的表为章节【执行建表语句->filesystem数据源】所建的表
  2. SELECT window_start, window_end, SUM(price) FROM TABLE
  3.  (TUMBLE(TABLE bid, DESCRIPTOR(bidtime), INTERVAL '10' MINUTES))
  4. GROUP BY window_start, window_end;

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/article/detail/43841
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号