看了这么多技术文章，你能理解作者想让你读完文章后学到什么吗？

大数据羊说__文章会让你明白

1. 博主会阐述了博主希望这篇文章能给朋友们带来什么帮助，让他们直观地理解博主的想法

2. 博主会从实际应用场景和案例入手，不仅仅是知识点的简单堆砌

3. 博主会分析重要知识点的原理，让小伙伴萌简单易行

gzh：大数据羊说

与大家一起学习大数据开发、面试、框架、实时离线计算、数仓保障 Flink/ClickHouse/Kafka/Spark/Hadoop 技术干货、数据下载、源代码解读等

52篇原创内容

后台回复源码微信官方账号1.13.2 最全 flink sql获取。

呕心沥血，Flink SQL 成神之路出品。小伙伴萌可以先体验一下下图大纲。由于微信公众号限制上传图片像素，所以博主分隔成了 5 图片。

NB

后台回复源码微信官方账号1.13.2 最全 flink sql获取。

Apache Flink 提供了两种关系型 API 统一流、批处理，Table 和 SQL API。

1. ? Table API 是一种集成 Java、Scala 和 Python 语言查询 API，简单的理解就是用 Java、Scala、Python 按照 SQL 查询接口封装一层 lambda 表达式查询 API，允许各种关系操作符(如选择、筛选和连接)的查询操作以强类型接口的形式组合，然后生成一个 Flink 任务运行。如下案例所示：

import org.apache.flink.table.api.*;  import static org.apache.flink.table.api.Expressions.*;  EnvironmentSettings settings = EnvironmentSettings     .newInstance()     .inStreamingMode()     .build();  TableEnvironment tEnv = TableEnvironment.create(settings);  // 下面就是 Table API 的案例，它的语义于 // select a, count(b) as cnt  // from Orders // group by a DataSet result = tEnv         .from("Orders")         .groupBy($("a"))         .select($("a"), $("b").count().as("cnt"))         .toDataSet(counts, Row.class);  result.print();  

2. ? SQL API 是基于 SQL 标准的 Apache Calcite 我们可以使用纯框架 SQL 开发和运营一个 Flink 任务如下：

insert into target select a, count(b) as cnt from Orders group by a  

注:无论输入是连续(流处理)还是有界(批处理)，在 Table 和 SQL 任一 API 中同一查询句语义相同，结果相同。这就是为什么 Flink SQL 和 Table API 在用户接口层面可以统一流批。xdm，用一套 SQL 既能跑任务，又能跑批任务，不香吗？

根据朋友使用的编程语言的不同（Java 或 Scala），项目中需要添加相应的依赖包。

1. ? Java 依赖如下

   org.apache.flink   flink-table-api-java-bridge_2.11   1.13.5      org.apache.flink   flink-table-planner-blink_2.11   1.13.5     org.apache.flink   flink-streaming-java_2.11   1.13.5     org.apache.flink   flink-table-common   1.13.5   

2. ? Scala 依赖如下

   org.apache.flink   flink-table-api-scala-bridge_2.11   1.13.5      org.apache.flink   flink-table-planner-blink_2.11   1.13.5   provided     org.apache.flink   flink-streaming-scala_2.11   1.13.5   provided     org.apache.flink   flink-table-common   1.13.5   

引入上述依赖后，可以开始使用小伙伴萌 Table\SQL API 是的。具体案例如下文所示。

在小伙伴萌看下面之前，先看看 2.2 节日的整体思路，跟随博主的思路，会更清晰:

1. ? 先通过一个 SQL\Table API 看看我们在实际开发中的代码结构应该是什么样子，让每个人都能有直观的感觉

2. ⭐ 重点介绍 SQL\Table API 中核心 API - TableEnvironment。SQL\Table 所有能用的接口都在 TableEnvironment 中

3. ⭐ 通过两个角度（外部表\视图、临时\非临时）认识 Flink SQL 体系中的表的概念

4. ⭐ 举几个创建外部表、视图的实际应用案例

// 创建一个 TableEnvironment，为后续使用 SQL 或者 Table API 提供上线
EnvironmentSettings settings = EnvironmentSettings
    .newInstance()
    .inStreamingMode() // 声明为流任务
    //.inBatchMode() // 声明为批任务
    .build();

TableEnvironment tEnv = TableEnvironment.create(settings);

// 创建一个输入表
tableEnv.executeSql("CREATE TEMPORARY TABLE table1 ... WITH ( 'connector' = ... )");
// 创建一个输出表
tableEnv.executeSql("CREATE TEMPORARY TABLE outputTable ... WITH ( 'connector' = ... )");

// 1. 使用 Table API 做一个查询并返回 Table
Table table2 = tableEnv.from("table1").select(...);
// 2. 使用 SQl API 做一个查询并返回 Table
Table table3 = tableEnv.sqlQuery("SELECT ... FROM table1 ... ");

// 将 table2 的结果使用 Table API 写入 outputTable 中，并返回结果
TableResult tableResult = table2.executeInsert("outputTable");
tableResult...

总结一下上面案例使用到的一些 API，让大家先对 Table\SQL API 的能力有一个大概了解：

1. ⭐ TableEnvironment：Table API 和 SQL API 的都集成在一个统一上下文（即 TableEnvironment）中，其地位等同于 DataStream API 中的 StreamExecutionEnvironment 的地位

2. ⭐ TableEnvironment::executeSql：用于 SQL API 中，可以执行一段完整 DDL，DML SQL。举例，方法入参可以是 CREATE TABLE xxx，INSERT INTO xxx SELECT xxx FROM xxx。

3. ⭐ TableEnvironment::from(xxx)：用于 Table API 中，可以以强类型接口的方式运行。方法入参是一个表名称。

4. ⭐ TableEnvironment::sqlQuery：用于 SQL API 中，可以执行一段查询 SQL，并把结果以 Table 的形式返回。举例，方法的入参是 SELECT xxx FROM xxx

5. ⭐ Table::executeInsert：用于将 Table 的结果插入到结果表中。方法入参是写入的目标表。

无论是对于 SQL API 来说还是对于 Table API 来说，都是使用 TableEnvironment 接口承载我们的业务查询逻辑的。只是在用户的使用接口的方式上有区别，以上述的 Java 代码为例，Table API 其实就是模拟 SQL 的查询方式封装了 Java 语言的 lambda 强类型 API，SQL 就是纯 SQL 查询。Table 和 SQL 很多时候都是掺杂在一起的，大家理解的时候就可以直接将 Table 和 SQL API 直接按照 SQL 进行理解，不用强行做特殊的区分。

而且博主推荐的话，直接上 SQL API 就行，其实 Table API 在企业实战中用的不是特别多。你说 Table API 方便吧，它确实比 DataStream API 方便，但是又比 SQL 复杂。一般生产使用不多。

注意：由于 Table 和 SQL API 基本上属于一回事，后续如果没有特别介绍的话，博主就直接按照 SQL API 进行介绍了。

如果 xdm 想直接上手运行一段 Flink SQL 的代码。

可以直接在公众号后台回复1.13.2 最全 flink sql获取源代码。所有的源码都开源到 github 上面了。里面包含了非常多的案例。可以直接拿来在本地运行的！！！肥肠的方便。

TableEnvironment 是使用 SQL API 永远都离不开的一个接口。其是 SQL API 使用的入口（上下文），就像是你要使用 Java DataStream API 去写一个 Flink 任务需要使用到 StreamExecutionEnvironment 一样。

可以认为 TableEnvironment 在 SQL API 中的地位和 StreamExecutionEnvironment 在 DataStream 中的地位是一样的，都是包含了一个 Flink 任务运行时的所有上下文环境信息。大家这样对比学习会比较好理解。

TableEnvironment 包含的功能如下：

1. ⭐ ️Catalog 管理：Catalog 可以理解为 Flink 的 MetaStore，类似 Hive MetaStore 对在 Hive 中的地位，关于 Flink Catalog 的详细内容后续进行介绍

2. ⭐ ️表管理：在 Catalog 中注册表

3. ⭐️ SQL 查询：（这 TMD 还用说，最基本的功能啊），就像 DataStream 中提供了 addSource、map、flatmap 等接口

4. ⭐ UDF 管理：注册用户定义（标量函数：一进一出、表函数：一进多出、聚合函数：多进一出）函数

5. ⭐️ UDF 扩展：加载可插拔 Module（Module 可以理解为 Flink 管理 UDF 的模块，是可插拔的，可以让小伙伴萌自定义 Module，去支持奇奇怪怪的 UDF 功能）

6. ⭐ DataStream 和 Table（Table\SQL API 的查询结果）之间进行转换：目前 1.13 版本的只有流任务支持，批任务不支持。1.14 支持批任务。

接下来介绍如何创建一个 TableEnvironment。案例为 Java。easy game。

1. ⭐ 方法 1：通过 EnvironmentSettings 创建 TableEnvironment

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

// 1. 就是设置一些环境信息
EnvironmentSettings settings = EnvironmentSettings
    .newInstance()
    .inStreamingMode() // 声明为流任务
    //.inBatchMode() // 声明为批任务
    .build();

// 2. 创建 TableEnvironment
TableEnvironment tEnv = TableEnvironment.create(settings);

在 1.13 版本中。

如果你是 inStreamingMode，则最终创建出来的 TableEnvironment 实例为 StreamTableEnvironmentImpl。

如果你是 inBatchMode，则最终创建出来的 TableEnvironment 实例为 TableEnvironmentImpl。

它两虽然都继承了 TableEnvironment 接口，但是 StreamTableEnvironmentImpl 支持的功能更多一些。大家可以直接去看看接口实验一下，这里就不进行详细介绍。

2. ⭐ 方法 2：通过已有的 StreamExecutionEnvironment 创建 TableEnvironment

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

一个表的全名（标识）会由三个部分组成：Catalog 名称.数据库名称.表名称。如果 Catalog 名称或者数据库名称没有指明，就会使用当前默认值 default。

举个例子，下面这个 SQL 创建的 Table 的全名为 default.default.table1。

tableEnv.executeSql("CREATE TEMPORARY TABLE table1 ... WITH ( 'connector' = ... )");

下面这个 SQL 创建的 Table 的全名为 default.mydatabase.table1。

tableEnv.executeSql("CREATE TEMPORARY TABLE mydatabase.table1 ... WITH ( 'connector' = ... )");

表可以是常规的（外部表 TABLE），也可以是虚拟的（视图 VIEW）。

1. ⭐ 外部表 TABLE：描述的是外部数据，例如文件（HDFS）、消息队列（Kafka）等。依然拿离线 Hive SQL 举个例子，离线中一个表指的是 Hive 表，也就是所说的外部数据。

2. ⭐ 视图 VIEW：从已经存在的表中创建，视图一般是一个 SQL 逻辑的查询结果。对比到离线的 Hive SQL 中，在离线的场景（Hive 表）中 VIEW 也都是从已有的表中去创建的。其实 Flink 也是一样的。

注意：

这里有不同的地方就是，离线 Hive MetaStore 中不会有 Catalog 这个概念，其标识都是 数据库.数据表。

表（视图、外部表）可以是临时的，并与单个 Flink session（可以理解为 Flink 任务运行一次就是一个 session）的生命周期绑定。

表（视图、外部表）也可以是永久的，并且对多个 Flink session 都生效。

1. ⭐ 临时表：通常保存于内存中并且仅在创建它们的 Flink session（可以理解为一次 Flink 任务的运行）持续期间存在。这些表对于其它 session（即其他 Flink 任务或非此次运行的 Flink 任务）是不可见的。因为这个表的元数据没有被持久化。如下案例：

-- 临时外部表
CREATE TEMPORARY TABLE source_table (
    user_id BIGINT,
    `name` STRING
) WITH (
  'connector' = 'user_defined',
  'format' = 'json',
  'class.name' = 'flink.examples.sql._03.source_sink.table.user_defined.UserDefinedSource'
);

-- 临时视图
CREATE TEMPORARY VIEW query_view as
SELECT
    *
FROM source_table
;

2. ⭐ 永久表：需要外部 Catalog（例如 Hive Metastore）来持久化表的元数据。一旦永久表被创建，它将对任何连接到这个 Catalog 的 Flink session 可见且持续存在，直至从 Catalog 中被明确删除。如下案例：

-- 永久外部表。需要外部 Catalog 持久化！！！
CREATE TABLE source_table (
    user_id BIGINT,
    `name` STRING
) WITH (
  'connector' = 'user_defined',
  'format' = 'json',
  'class.name' = 'flink.examples.sql._03.source_sink.table.user_defined.UserDefinedSource'
);

-- 永久视图。需要外部 Catalog 持久化！！！
CREATE VIEW query_view as
SELECT
    *
FROM source_table
;

3. ⭐ 如果临时表和永久表使用了相同的名称（Catalog名.数据库名.表名）。那么在这个 Flink session 中，你的任务访问到这个表时，访问到的永远是临时表（即相同名称的表，临时表会屏蔽永久表）。

由于目前在实时数据的场景中多以消息队列作为数据表。此处就以 Kafka 为例创建一个外部数据表。

1. ⭐ Table API 创建外部数据表

public static void main(String[] args) throws Exception {

    StreamExecutionEnvironment env =
            StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
    
    EnvironmentSettings settings = EnvironmentSettings
            .newInstance()
            .useBlinkPlanner()
            .inStreamingMode()
            .build();

    StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings);

    // kafka 数据源
    DataStream r = env.addSource(new FlinkKafkaConsumer(xxx));
    // 将 DataStream 转为一个 Table API 中的 Table 对象进行使用
    Table sourceTable = tEnv.fromDataStream(r
            , Schema
                    .newBuilder()
                    .column("f0", "string")
                    .column("f1", "string")
                    .column("f2", "bigint")
                    .columnByExpression("proctime", "PROCTIME()")
                    .build());

    tEnv.createTemporaryView("source_table", sourceTable);

    String selectWhereSql = "select f0 from source_table where f1 = 'b'";

    Table resultTable = tEnv.sqlQuery(selectWhereSql);

    tEnv.toRetractStream(resultTable, Row.class).print();

    env.execute();
}

上述案例中，Table API 将一个 DataStream 的结果集通过 StreamTableEnvironment::fromDataStream 转为一个 Table 对象来使用。

2. ⭐ SQL API 创建外部数据表

EnvironmentSettings settings = EnvironmentSettings
        .newInstance()
        .useBlinkPlanner()
        .inStreamingMode()
        .build();

StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings);

// SQL API 执行 create table 创建表
tEnv.executeSql(
        "CREATE TABLE KafkaSourceTable (\n"
                + "  `f0` STRING,\n"
                + "  `f1` STRING\n"
                + ") WITH (\n"
                + "  'connector' = 'kafka',\n"
                + "  'topic' = 'topic',\n"
                + "  'properties.bootstrap.servers' = 'localhost:9092',\n"
                + "  'properties.group.id' = 'testGroup',\n"
                + "  'format' = 'json'\n"
                + ")"
);

Table t = tEnv.sqlQuery("SELECT * FROM KafkaSourceTable");

具体的创建方式就是使用 Create Table xxx DDL 定义一个 Kafka 数据源（输入）表（也可以是 Kafka 数据汇（输出）表）。

9

xdm，是不是又和 Hive 一样？惊不惊喜意不意外。对比学习 +1。

上文已经说了，一个 VIEW 其实就是一段 SQL 逻辑的查询结果。

视图 VIEW 在 Table API 中的体现就是：一个 Table 的 Java 对象，其封装了一段查询逻辑。如下案例所示：

1. ⭐ Table API 创建 VIEW

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

EnvironmentSettings settings = EnvironmentSettings
    .newInstance()
    .inStreamingMode() // 声明为流任务
    .build();

TableEnvironment tEnv = TableEnvironment.create(settings);

// Table API 中的一个 Table 对象
Table projTable = tEnv.from("X").select(...);

// 将 projTable 创建为一个叫做 projectedTable 的 VIEW
tEnv.createTemporaryView("projectedTable", projTable);

Table API 是使用了 TableEnvironment::createTemporaryView 接口将一个 Table 对象创建为一个 VIEW。

2. ⭐ SQL API 创建 VIEW

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

EnvironmentSettings settings = EnvironmentSettings
    .newInstance()
    .inStreamingMode() // 声明为流任务
    .build();

TableEnvironment tEnv = TableEnvironment.create(settings);

String sql = "CREATE TABLE source_table (\n"
    + "    user_id BIGINT,\n"
    + "    `name` STRING\n"
    + ") WITH (\n"
    + "  'connector' = 'user_defined',\n"
    + "  'format' = 'json',\n"
    + "  'class.name' = 'flink.examples.sql._03.source_sink.table.user_defined.UserDefinedSource'\n"
    + ");\n"
    + "\n"
    + "CREATE TABLE sink_table (\n"
    + "    user_id BIGINT,\n"
    + "    name STRING\n"
    + ") WITH (\n"
    + "  'connector' = 'print'\n"
    + ");\n"
    + "CREATE VIEW query_view as\n" // 创建 VIEW
    + "SELECT\n"
    + "    *\n"
    + "FROM source_table\n"
    + ";\n"
    + "INSERT INTO sink_table\n"
    + "SELECT\n"
    + "    *\n"
    + "FROM query_view;";

Arrays.stream(sql.split(";"))
      .forEach(tEnv::executeSql);

SQL API 是直接通过一段 CREATE VIEW query_view as select * from source_table 来创建的 VIEW，是纯 SQL 写法。

9

这种创建方式是不是贼熟悉，和离线 Hive 一样 +1~

注意：

在 Table API 中的一个 Table 对象被后续的多个查询使用的场景下：Table 对象不会真的产生一个中间表供下游多个查询去引用，即多个查询不共享这个 Table 的结果，小伙伴萌可以理解为是一种中间表的简化写法，不会先产出一个中间表结果，然后将这个结果在下游多个查询中复用，后续的多个查询会将这个 Table 的逻辑执行多次。类似于 with tmp as (DML) 的语法

首先，如果 xdm 想直接上手运行一段 Flink SQL 的代码。

可以直接在公众号后台回复1.13.2 最全 flink sql获取源代码。所有的源码都开源到 github 上面了。里面包含了非常多的案例。可以直接拿来在本地运行的！！！肥肠的方便。

来看看一个 SQL 查询案例。

1. ⭐ 案例场景：计算每一种商品（sku_id 唯一标识）的售出个数、总销售额、平均销售额、最低价、最高价

2. ⭐ 数据准备：数据源为商品的销售流水（sku_id：商品，price：销售价格），然后写入到 Kafka 的指定 topic（sku_id：商品，count_result：售出个数、sum_result：总销售额、avg_result：平均销售额、min_result：最低价、max_result：最高价）当中

3. ⭐ 任务代码：

EnvironmentSettings settings = EnvironmentSettings
    .newInstance()
    .inStreamingMode() // 声明为流任务
    //.inBatchMode() // 声明为批任务
    .build();

TableEnvironment tEnv = TableEnvironment.create(settings);

// 1. 创建一个数据源（输入）表，这里的数据源是 flink 自带的一个随机 mock 数据的数据源。
String sourceSql = "CREATE TABLE source_table (\n"
        + "    sku_id STRING,\n"
        + "    price BIGINT\n"
        + ") WITH (\n"
        + "  'connector' = 'datagen',\n"
        + "  'rows-per-second' = '1',\n"
        + "  'fields.sku_id.length' = '1',\n"
        + "  'fields.price.min' = '1',\n"
        + "  'fields.price.max' = '1000000'\n"
        + ")";

// 2. 创建一个数据汇（输出）表，输出到 kafka 中
String sinkSql = "CREATE TABLE sink_table (\n"
        + "    sku_id STRING,\n"
        + "    count_result BIGINT,\n"
        + "    sum_result BIGINT,\n"
        + "    avg_result DOUBLE,\n"
        + "    min_result BIGINT,\n"
        + "    max_result BIGINT,\n"
        + "    PRIMARY KEY (`sku_id`) NOT ENFORCED\n"
        + ") WITH (\n"
        + "  'connector' = 'upsert-kafka',\n"
        + "  'topic' = 'tuzisir',\n"
        + "  'properties.bootstrap.servers' = 'localhost:9092',\n"
        + "  'key.format' = 'json',\n"
        + "  'value.format' = 'json'\n"
        + ")";

// 3. 执行一段 group by 的聚合 SQL 查询
String selectWhereSql = "insert into sink_table\n"
        + "select sku_id,\n"
        + "       count(*) as count_result,\n"
        + "       sum(price) as sum_result,\n"
        + "       avg(price) as avg_result,\n"
        + "       min(price) as min_result,\n"
        + "       max(price) as max_result\n"
        + "from source_table\n"
        + "group by sku_id";

tEnv.executeSql(sourceSql);
tEnv.executeSql(sinkSql);
tEnv.executeSql(selectWhereSql);

大家会比较好奇，要写 SQL 就纯 SQL 呗，要写 DataStream 就纯 DataStream 呗，为啥还要把这两类接口做集成呢？

博主举一个案例：在 pdd 这种发补贴券的场景下，希望可以在发的补贴券总金额超过 1w 元时，及时报警出来，来帮助控制预算，防止发的太多。

对应的解决方案，我们可以想到使用 SQL 计算补贴券发放的结果，但是 SQL 的问题在于无法做到报警。所以我们可以将 SQL 的查询的结果（即 Table 对象）转为 DataStream，然后就可以在 DataStream 后自定义报警逻辑的算子。

我们直接上 SQL 和 DataStream API 互相转化的案例：

public static void main(String[] args) throws Exception {

    FlinkEnv flinkEnv = FlinkEnvUtils.getStreamTableEnv(args);

    // 1. pdd 发补贴券流水数据
    String createTableSql = "CREATE TABLE source_table (\n"
            + "    id BIGINT,\n" -- 补贴券的流水 id
            + "    money BIGINT,\n" -- 补贴券的金额
            + "    row_time AS cast(CURRENT_TIMESTAMP as timestamp_LTZ(3)),\n"
            + "    WATERMARK FOR row_time AS row_time - INTERVAL '5' SECOND\n"
            + ") WITH (\n"
            + "  'connector' = 'datagen',\n"
            + "  'rows-per-second' = '1',\n"
            + "  'fields.id.min' = '1',\n"
            + "  'fields.id.max' = '100000',\n"
            + "  'fields.money.min' = '1',\n"
            + "  'fields.money.max' = '100000'\n"
            + ")\n";

    // 2. 计算总计发放补贴券的金额
    String querySql = "SELECT UNIX_TIMESTAMP(CAST(window_end AS STRING)) * 1000 as window_end, \n"
            + "      window_start, \n"
            + "      sum(money) as sum_money,\n" -- 补贴券的发放总金额
            + "      count(distinct id) as count_distinct_id\n"
            + "FROM TABLE(CUMULATE(\n"
            + "         TABLE source_table\n"
            + "         , DESCRIPTOR(row_time)\n"
            + "         , INTERVAL '5' SECOND\n"
            + "         , INTERVAL '1' DAY))\n"
            + "GROUP BY window_start, \n"
            + "        window_end";

    flinkEnv.streamTEnv().executeSql(createTableSql);

    Table resultTable = flinkEnv.streamTEnv().sqlQuery(querySql);

    // 3. 将金额结果转为 DataStream，然后自定义超过 1w 的报警逻辑
    flinkEnv.streamTEnv()
            .toDataStream(resultTable, Row.class)
            .flatMap(new FlatMapFunction() {
                @Override
                public void flatMap(Row value, Collector