FlinkTable＆SQL(六)

FlinkTable&SQL(六、七)

今日目标

• 了解Flink Table&SQL发展历史
• 了解为什么要使用Table API & SQL
• 掌握Flink Table&SQL进行批处理开发
• 掌握Flink Table&SQL进行流处理开发
• 掌握常用的开发案例
• Flink-SQL的常用算子

Flink Table & SQL

• FlinkTable & SQL 是抽象级别更高的操作， 底层Flink Runtime => Stream 流程

• 批处理是流处理的一种特殊形态

• FlinkSQL 遵循ANSI的SQL规范

• Flink1.9之前， FlinkSQL包括两套Table api ， DataStream Table API（流处理） ，DataSet Table API（批处理）

• Planner 查询器， 抽象语法树，parser、optimizer、codegen（模板代码生成），最终生成 Flink Runtime 直接进行执行的代码

• Planner包括old Planner 和 Blink Planner ，Blink Planner 底层实现了 流批一体（默认的Planner）

FlinkTable & SQL 程序结构

• 导入 pom 依赖， jar包坐标

  <dependency>            <groupId>org.apache.flink</groupId>            <artifactId>flink-table-api-scala-bridge_2.12</artifactId>            <version>${flink.version}</version>        </dependency>        <dependency>            <groupId>org.apache.flink</groupId>            <artifactId>flink-table-api-java-bridge_2.12</artifactId>            <version>${flink.version}</version>        </dependency>        <!-- flink执行计划,这是1.9版本之前的-->        <dependency>            <groupId>org.apache.flink</groupId>            <artifactId>flink-table-planner_2.12</artifactId>            <version>${flink.version}</version>        </dependency>        <!-- blink执行计划,1.11+默认的-->        <dependency>            <groupId>org.apache.flink</groupId>            <artifactId>flink-table-planner-blink_2.12</artifactId>            <version>${flink.version}</version>        </dependency>        <dependency>            <groupId>org.apache.flink</groupId>            <artifactId>flink-table-common</artifactId>            <version>${flink.version}</version>        </dependency>

• 

• 创建 FlinkTable FlinkSQL的 表的方式

  // table is the result of a simple projection query Table projTable = tableEnv.from("X").select(...);// register the Table projTable as table "projectedTable"tableEnv.createTemporaryView("projectedTable", projTable);

• SQL的四种语句

  1. DDL 数据定义语言， 创建数据库、表，删除数据库、表
  2. DML 数据操作语言， 对数据进行增、删、改操作
  3. DCL 数据控制语言， 对数据的操作权限进行设置 grant revoke
  4. DQL 数据查询语言，对数据表中的数据进行查询，基础查询，复杂查询，多表查询，子查询

• 需求

  将两个数据流 DataStream 通过 FlinkTable & SQL API 进行 union all 操作，条件ds1 amount>2 union all ds2 amount<2

• 开发步骤

  package cn.itcast.flink.sql;import lombok.AllArgsConstructor;import lombok.Data;import lombok.NoArgsConstructor;import org.apache.flink.streaming.api.datastream.DataStream;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.table.api.EnvironmentSettings;import org.apache.flink.table.api.Table;import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;import org.apache.flink.types.Row;import java.util.Arrays;import static org.apache.flink.table.api.Expressions.$;/** * Author itcast * Date 2021/6/22 9:45 * Desc TODO */public class FlinkTableAPIDemo {    public static void main(String[] args) throws Exception {        //1.准备环境 创建流环境 和 流表环境，并行度设置为1        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        env.setParallelism(1);        EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();        //创建流表环境        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env,settings);        //2.Source 创建数据集        DataStream<Order> orderA = env.fromCollection(Arrays.asList(                new Order(1L, "beer", 3),                new Order(1L, "diaper", 4),                new Order(3L, "rubber", 2)));        DataStream<Order> orderB = env.fromCollection(Arrays.asList(                new Order(2L, "pen", 3),                new Order(2L, "rubber", 3),                new Order(4L, "beer", 1)));        //3.注册表 将数据流转换成表        // 通过fromDataStream将数据流转换成表        Table orderTableA = tEnv.fromDataStream(orderA, $("user"), $("product"), $("amount"));        // 将数据流转换成 创建临时视图        tEnv.createTemporaryView("orderTableB",orderB,$("user"), $("product"), $("amount"));        //4.执行查询，查询order1的amount>2并union all 上 order2的amoun<2的数据生成表        Table result = tEnv.sqlQuery("" +                "select * from " + orderTableA + " where amount>2 " +                "union all " +                "select * from orderTableB where amount<2");        //4.1 将结果表转换成toAppendStream数据流        //字段的名称和类型        result.printSchema();        DataStream<Row> resultDS = tEnv.toAppendStream(result, Row.class);        //5.打印结果        resultDS.print();        //6.执行环境        env.execute();        // 创建实体类 user:Long product:String amount:int    }    @Data    @NoArgsConstructor    @AllArgsConstructor    public static class Order {        public Long user;        public String product;        public int amount;    }}