Flink学习4-流式SQL

约定不等于承诺〃 2024-04-20 08:13 92阅读 0赞

# Flink学习4-流式SQL #

# Flink系列文章 #

*  更多Flink系列文章请点击[Flink系列文章][Flink]
 *  更多大数据文章请点击[大数据好文推荐][Link 1]

# 摘要 #

介绍Flink Table Sql API相关概念，还会提供一些例子。

本文大部分内容已经更新到Flink 1.11。

# 1 基本概念 #

## 1.1 概述 ##

Flink高层API有两种：Table级别和SQL级别。两种API都是统一的处理批和流数据，也就是说对于无界、实时的流或者有界、记录型的流有着同样的处理语义，产生同样的结果。

**Table和SQL API采用了`Apache Calcite`进行语句解析、验证和查询调优。**  
他们可以和DataStream及DataSet API无缝集成，并支持用户自定义的标量，聚合和表值函数。

Flink的关系型API旨在简化数据分析，数据管道和ETL应用程序。

下面这个示例功能和DataStream API中的相同，也是展示一个SQL查询将一个点击流session化，然后对每个session中的点击数计数：

SELECT userId, COUNT(*)
    FROM clicks
    GROUP BY SESSION(clicktime, INTERVAL '30' MINUTE), userId

这个SQL就是个流式处理SQL，简洁，高效。

## 1.2 限制 ##

虽然flink 1.9.0支持DDL，但是尚不支持Time相关的元素。

可见:

*  [FLIP-66: Support Time Attribute in SQL DDL][FLIP-66_ Support Time Attribute in SQL DDL]
 *  [JIRA-Support Time Attribute in SQL DDL][]

**注意，1.10开始版本中DDL已经支持Time相关内容！**

## 1.3 动态表(Dynamic Tables) ##

### 1.3.1 概念 ###

*  动态表主要是对应了传统有界数据概念上的表，对比如下：
    
    <table> 
     <thead> 
      <tr> 
       <th>/</th> 
       <th>关系代数/SQL</th> 
       <th>流处理</th> 
      </tr> 
     </thead> 
     <tbody> 
      <tr> 
       <td>数据量</td> 
       <td>有界</td> 
       <td>无界</td> 
      </tr> 
      <tr> 
       <td>访问区域</td> 
       <td>查询时可访问完整数据</td> 
       <td>流式查询时不能访问完整数据，而是不断对新输入的数据访问并计算</td> 
      </tr> 
      <tr> 
       <td>执行周期</td> 
       <td>执行一段时间后完毕</td> 
       <td>用不完结，持续执行，并根据收到的数据不断更新计算结果</td> 
      </tr> 
     </tbody> 
    </table>
 *  物化视图(Materialized Views)与流SQL  
    高级数据库用来应对流数据场景，他被定义为一条SQL查询，会缓存查询结果用来查询访问（与虚拟视图需要执行SQL不同，无需每次使用都执行查询计算）。有一种`Eager View Maintenance`技术被用来在基表（定义视图时的查询语句的原表）更新时更新物化视图。
    
    我们可以想象，数据库表一般拥有一个变化的由增删改组成的数据流（Changelog Stream），那就需要持续查询处理该数据流来不断更新物化视图（物化视图本身就由SQL查询定义），也就是说物化视图就是流式SQL查询的结果。
 *  动态表和持续查询（Continuous Query）  
    有了以上物化视图的相关概念，这里我们引出动态表。动态表是Flink 流式Table和SQL的核心概念（一般是逻辑概念，并不一定要在查询执行中物化），会持续更新。
    
    需要注意的是，Flink可以像查询静态表一样查询它们。
    
    动态表可像传统表一样执行查询（称为`持续查询-Continuous Query`），这种持续查询的结果形成动态表，并且不断更新该动态表。这跟更新物化视图差不多，持续查询在语义上也和批量模式中查询任意时刻的输入表的快照相同。这个过程如下图：  
    ![在这里插入图片描述][20200415200701974.png]
    
    1.  Input Stream被转为动态表
    2.  持续查询在动态表上执行，生成新的动态表
    3.  生成的新动态表又被转回Output Stream
 *  动态表的存储  
    Flink中动态表只是一个逻辑概念，Table和SQL API用它来统一处理有界和无界数据，注意Flink是不会保存动态表数据的，而是存储在外部系统之中（如数据库、MQ、文件系统等）。

### 1.3.2 例子 ###

#### 1.3.2.1 背景 ####

比如一个点击流事件数据流，schema如下：

[
      user:  VARCHAR,   // 用户名
      cTime: TIMESTAMP, // 访问 URL 的时间
      url:   VARCHAR    // 用户访问的 URL
    ]

#### 1.3.2.2 Group Aggregation ####

点击事件流随着事件不断流入，其实就相当于不断地对结果表的`INSERT`操作，可在逻辑上构成动态表（这种在流上定会以的表，其实在Flink内没有物化）。也就是说，本质上Flink使用仅有`INSERT`的Chanelog输入流来构建动态表。

下图左侧为点击流事件，右侧为转换后的动态表，随着点击流持续Insert而不断增长  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70]  
首先考虑以下sql，计算用户维度的url访问总量。

select user,count(url) as cnt from clicks group by user;

此时每条数据都会触发聚合计算，一般需要接入能更新的数据源，如果是不能更新的如kafka就是一条条单独数据，持续查询原表生成动态表如下图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70 1]

1.  查询刚开始时，输入表（左表）为空
2.  当第一行`[Mary, ./home]`输入clicks原表时，输出了首个结果(INSERT)
3.  第二行输入时，计算后INSERT结果到动态表
4.  第三行又来一个Mary相关的，所以对动态表中的`[Mary, 1]`执行了UPDATE操作，更新为`[Mary, 2]`
5.  第四行记录也是新人，所以也是INSERT到动态表

#### 1.3.2.3 Window Aggregation ####

再来一个复杂一点的sql，计算每个1小时的翻滚窗口内（以事件时间计算）的用户维度的url访问总量。

select user,TUMBLE_END(cTime, INTERVAL '1' HOURS) as endT,count(ur) as cnt from clicks 
    group by user,TUMBLE(cTime, INTERVAL '1' HOURS);

此时数据会累积，直到水位到达窗口边界才会触发上一窗口计算，持续查询原表生成动态表如下图，这里颜色相同的为同一个user：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70 2]

*  根据cTime列不同，划分为不同窗口，分别进行计算。
 *  每次计算后将结果INSERT到动态表

#### 1.3.2.4 Window Aggregation对比Group Aggregation ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70 3]  
`Group Aggregation`如果不配置会导致结果表无限增大，最终会出问题。所以一般使用该模式需要清理掉过期数据并设置最小、最大时间。

StreamTableEnvironment.create(bsEnv, bsSettings).getConfig.setIdleStateRetentionTime(Time.days(1), Time.days(2))

这个用来指定空闲State被保留的的最小和最大时长，差值至少5分钟。超过最大时长未更新的State会被移除。

State清理后，关于该已经被清理的State的新来的数据会被当做首条数据处理，可能导致覆盖之前的结果。

### 1.3.3 Update / Append ###

虽然以上例子都是聚合运算，但有很大不同：

*  普通聚合运算，是不断`UPDATE`结果表，用来定义结果表的Changelog流中包含`INSERT`和`UPDATE`。
 *  窗口聚合运算的Changelog流仅包含`INSERT`，只能`Append`到结果表！

一个查询是`append-only`还是`update`的影响：

*  update表通常需要维护更多状态，用来更新
 *  两种表转为Stream的方式不同

### 1.3.4 表到流的转换 ###

#### 1.3.4.1 概述 ####

动态表支持`INSERT` 、`UPDATE`、`DELETE`操作来不断修改。所以动态表可能是：

*  只有不断被修改的单行数据
 *  一个`insert-only`表，没有`UPDATE` / `DELETE`来修改表
 *  或者介于以上两者之间

当要将动态表转为流或写入外部系统时，需要将将这三种操作编码。

需要注意的是，只有`Append`和`Retract`流模式下，才可以将动态表转为DataStream。详见table2datastream

使用`TableSink`接口来讲动态表发送到外部系统相关内容可见[TableSources and TableSinks][]

还可参考

*  [Flink Table 的三种 Sink 模式][Flink Table _ Sink]

#### 1.3.4.2 Append Only Mode（增） ####

仅交互`INSERT`操作数据。

此时可通过将insert的新记录发送到下游来转换为流。

比如我们前面的例子2，翻滚窗口计算后的变化数据流中仅有INSERT变化，全部append到结果动态表中。

#### 1.3.4.3 Retract Mode （增删改） ####

编码：  
交互`INSERT`（编码为`ADD Message`）和`DELETE`（编码为`RETRACT Message`）、`UPDATE`（对于修改前的行来说编码为`RETRACT Message`，对于正在修改的行来说编码为`ADD Message`）操作数据。

特点：

*  与Upsert Mode相反，Retract Mode不需要定义key。
 *  每个UPDATE操作由两条消息（RETRACT和ADD）组成，效率较低。

将动态表转为Retract Stream的例子：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70 4]  
上图解读：

*  数据到来的顺序从上往下
 *  第三行的Mary是第二次读到的Mary，此时由于第一行已经读到过Mary（`+ Mary,1`），所以需要更新。  
    此时需要先DELETE之前的`- Mary,1`，再INSERT `+ Mary,2`，即upsert更新由两个操作实现。

#### 1.3.4.4 Upsert Mode（增删改） ####

编码：  
交互`INSERT`和`UPDATE`（编码为`UPSERT Message`）和`DELETE`（编码为`Delete Message`）操作数据。

特点：

*  Upsert Mode需要一个唯一的key（可能是组合的），用来传播UPDATE事件。  
    具体来说，**外部连接器需要了解该唯一key属性，才能正确处理消息。**
 *  Upsert Mode和Retract Mode都支持增删改但编码不同  
    Upsert Mode中的UPDATE事件使用单条UPSERT消息进行编码，而Retract Mode中的UPDATE由两条消息（RETRACT和ADD）组成，因此Upsert模式效率更高。

例子：  
比如我们前面的例子1，计算后的变化数据流中有INSERT和UPDATE变化，会持续更新动态表中之前的结果。此时比起Append Only Mode来说会需要维护更多的状态信息，用来更新。但相比Retract Mode，本模式在修改数据时可直接Upsert，无需两个操作即可完成。

将动态表转为Upsert Stream的示意图如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70 5]

*  比如第三行的第二次遇到Mary，就直接修改即完成。
 *  注意看，上面的`user`列就是唯一KEY！

### 1.3.5 动态表查询限制 ###

许多(但不是全部)语义上有效的查询可以作为流上持续查询进行代价评估。有些查询代价太高而无法计算，这可能是由于它们需要维护的状态大小，也可能是由于计算以及更新代价太高：

*  状态大小  
    持续运行的流式应用，有的需要更新之前输出的结果，那就必须维护所有输出行。
    
    比如前面的例子1就需要一直维护每个用户的URL总和以便每次计算时更新。随着时间流逝，注册用户越来越多，有的网站甚至会为未注册用户也绑定一个用户名，导致维护的状态会越来越大，最终可能导致这个持续查询失败！
 *  计算和更新  
    一些持续查询需要重新计算和更新大量结果行，甚至即使只添加或更新了一行数据。这种情况显然不适合持续查询。
    
    比如以下SQL，以每个用户最后一次点击时间来为每个用户计算RANK：
    
        SELECT user, RANK() OVER (ORDER BY lastLogin)
        	FROM (
        	  SELECT user, MAX(cTime) AS lastAction FROM clicks GROUP BY user
        	);
        	```
        此时一旦clicks表接收到一行新数据，则用户的`lastAction`字段就会更新，而新的rank也需要被计算。这会导致所有人的排名都重新计算！有一些参数可以控制维持状态大小和结果准确度之间的权衡：[Query Configuration](https://ci.apache.org/projects/flink/flink-docs-master/dev/table/streaming/query_configuration.html)

## 1.4 Catalog ##

可参考

*  [Catalogs][]
 *  [HiveCatalog][]

### 1.4.1 简介 ###

Catalog用来存储元数据，包括数据库、表、分区、视图、函数以及需要访问数据库或外部系统数据的信息。

Catalog提供的元数据包括两方面：

*  临时元数据  
    如临时表、通过 TableEnvironment 注册的临时UDF等
 *  持久化元数据  
    如存在HiveMetastore中的数据

Catalog提供统一的Table API和SQL来管理、访问元数据。

### 1.4.2 元数据映射 ###

Catalog使得用户可直引用其他数据系统的元数据，映射到Flink。比如直接将JDBC表映射到Flink表，不需要重写FlinkSql DDL就能直接查询外部表了，十分方便。

### 1.4.3 可用的Catalog ###

#### 1.4.3.1 概述 ####

目前有

*  GenericInMemoryCatalog  
    基于内存，所有元数据只在 session 的生命周期内可用。
    
    注意，大小写敏感。
 *  JdbcCatalog  
    使用JDBC协议连接管理元数据，目前只有`PostgresCatalog`实现。
    
    相见[JdbcCatalog documentation][]
 *  HiveCatalog  
    可用来拿出Flink元数据、作为接口读写Hive现有元数据。
    
    注意，存储时都是小写。
 *  [自定义Catalog][Catalog]

#### 1.4.3.2 HiveCatalog ####

可参考

*  [HiveCatalog][]
 *  [Hive集成方法][Hive]

可以让Flink使用Hive Catalog存储Flink SQL 元数据。

可以在Hive命令行中使用`DESCRIBE FORMATTED`命令查看表的元数据，如果是`is_generic=true`代表是Flink专用表，这种表只能由Flink读写使用，不要用Hive去读写。

也可以直接使用Flink读写Hive表数据。

需要将以下包放入`$FLINK_HOME/lib`:

*  flink-connector-hive\_2.11-1.11.0.jar
 *  hive-exec-2.3.3.jar

使用hive存储flink sql元数据的时候，还可以做一些高级配置，可参考[Flink-hive\_catalog][Flink-hive_catalog]

简单来说，可以修改`$FLINK_HOME/conf/sql-client-defaults.yaml`:

execution:
        planner: blink
        type: streaming
        ...
        current-catalog: zeppelin-hive  # set the HiveCatalog as the current catalog of the session
        current-database: flink_meta
        
    catalogs:
       - name: myhive
         type: hive
         hive-conf-dir: /opt/hive-conf  # contains hive-site.xml
         hive-version: 2.3.3

然后可以启动`sql-client`，会默认读取`$FLINK_HOME/conf/sql-client-defaults.yaml`：

sql-client.sh embedded

如果报错`flink sql-client ClassNotFoundException: LogFactory`，则还需要设置`HADOOP_CLASSPATH`。

出现如下界面表示sql-client启动成功：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70 6]  
关于sql-client，可参考

*  [Flink 与 Hive 的磨合期][Flink _ Hive]
 *  [SQL Client Beta][]
 *  [Flink SQL Client初探][Flink SQL Client]

### 1.4.4 Catalog注册和创建 ###

#### 1.4.4.1 SQL ####

*  创建和注册  
    默认有内存中的default\_catalog，包括一个default\_database

// Create a HiveCatalog 
    val catalog = new HiveCatalog("myhive", null, "<path_of_hive_conf>", "<hive_version>")
    
    // Register the catalog
    tableEnv.registerCatalog("myhive", catalog)
    
    // Create a catalog database
    tableEnv.executeSql("CREATE DATABASE mydb WITH (...)")

*  切换当前catalog和数据库  
    Flink总是通过当前catalog和数据库来搜索表、视图、UDF，可切换使用的catalog和数据库

USE CATALOG myCatalog;
    USE myDB;

tableEnv.useCatalog("myCatalog");
    tableEnv.useDatabase("myDb");

*  查询非当前catalog的表

SELECT * FROM not_the_current_catalog.not_the_current_db.my_table;

*  创建表

tableEnv.executeSql("CREATE TABLE mytable (name STRING, age INT) WITH (...)")

*  展示catalog

show catalogs;

tableEnv.listCatalogs();

*  展示数据库

show databases;

tableEnv.listDatabases();

*  展示表

show tables;

// should return the tables in current catalog and database.
    tableEnv.listTables()

也可以用`yaml`来配置catalog，再执行sql.

#### 1.4.4.2 TableAPI ####

import org.apache.flink.table.api._
    import org.apache.flink.table.catalog._
    import org.apache.flink.table.catalog.hive.HiveCatalog
    import org.apache.flink.table.descriptors.Kafka
    
    val tableEnv = TableEnvironment.create(EnvironmentSettings.newInstance.build)
    
    // Create a HiveCatalog 
    val catalog = new HiveCatalog("myhive", null, "<path_of_hive_conf>", "<hive_version>")
    
    // Register the catalog
    tableEnv.registerCatalog("myhive", catalog)
    
    // Create a catalog database 
    catalog.createDatabase("mydb", new CatalogDatabaseImpl(...))
    
    // Create a catalog table
    val schema = TableSchema.builder()
        .field("name", DataTypes.STRING())
        .field("age", DataTypes.INT())
        .build()
    
    catalog.createTable(
            new ObjectPath("mydb", "mytable"), 
            new CatalogTableImpl(
                schema,
                new Kafka()
                    .version("0.11")
                    ....
                    .startFromEarlist()
                    .toProperties(),
                "my comment"
            ),
            false
        )
        
    val tables = catalog.listTables("mydb") // tables should contain "mytable"

# 2 Table API #

参考：

*  [Table API][]

Table&Sql API可参考[Concepts & Common API][Concepts _ Common API]

## 2.1 概述 ##

Table&Sql API核心在于`Table`，作为输入和输出。

## 2.2 Scala Api示例 ##

### 2.2.1 基本代码结构 ###

// create a TableEnvironment for specific planner batch or streaming
    val tableEnv = ... // see "Create a TableEnvironment" section
    
    // create a Table
    tableEnv.connect(...).createTemporaryTable("table1")
    // register an output Table
    tableEnv.connect(...).createTemporaryTable("outputTable")
    
    // create a Table from a Table API query
    val tapiResult = tableEnv.from("table1").select(...)
    // create a Table from a SQL query
    val sqlResult  = tableEnv.sqlQuery("SELECT ... FROM table1 ...")
    
    // emit a Table API result Table to a TableSink, same for SQL result
    val tableResult = tapiResult.executeInsert("outputTable")
    tableResult...

### 2.2.2 Blink例子 ###

批流统一的代码结构。

以下是使用Blink时的scala Streaming示例：

// **********************
    // BLINK STREAMING QUERY
    // **********************
    import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
    import org.apache.flink.table.api.EnvironmentSettings
    import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
    
    // create a TableEnvironment for blink planner streaming
    val bsEnv = StreamExecutionEnvironment.getExecutionEnvironment
    val bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()
    // 每个表都绑定到一个特定的TableEnvironment，不能跨TableEnvironment连接表，如join/union
    val tableEnv = StreamTableEnvironment.create(bsEnv, bsSettings)
    
    // 1. create a TableEnvironment for blink planner batch
    val bbSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build()
    val bbTableEnv = TableEnvironment.create(bbSettings)
    
    // 2. create Table 
    
    tableEnv.connect(...).createTemporaryTable("table1")
    // register an output Table
    tableEnv.connect(...).createTemporaryTable("outputTable")
    
    // create a Table from a Table API query
    val tapiResult = tableEnv.from("table1").select(...)
    // create a Table from a SQL query
    val sqlResult  = tableEnv.sqlQuery("SELECT ... FROM table1 ...")
    
    // emit a Table API result Table to a TableSink, same for SQL result
    val tableResult = tapiResult.executeInsert("outputTable")
    tableResult...

使用Blink时的scala Batch示例：

// ******************
    // BLINK BATCH QUERY
    // ******************
    import org.apache.flink.table.api.{ EnvironmentSettings, TableEnvironment}
    
    val bbSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build()
    val bbTableEnv = TableEnvironment.create(bbSettings)

## 2.3 TableEnvironment ##

### 2.3.1 概述 ###

TableEnvironment职责如下：

*  注册table到内部catalog
 *  注册catalog
 *  加载插件化的module
 *  执行sql查询
 *  注册UDF
 *  将DtaStream/DataSet转为Table
 *  持有指向ExecutionEnvironment或StreamExecutionEnvironment的引用

每个Table都会绑定到某个TableEnvironment，不能在同一个查询里跨TableEnvironment执行连接表操作（如join/union）。

### 2.3.2 创建和配置 ###

// create a TableEnvironment for blink planner streaming
    val bsEnv = StreamExecutionEnvironment.getExecutionEnvironment
    // TableConfig，可用来配置TableEnvironment、优化查询
    val bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()
    // TableEnvironment
    val tableEnv = StreamTableEnvironment.create(bsEnv, bsSettings)

## 2.4 与DataStream/DataSet集成 ##

### 2.4.1 概述 ###

参见[Integration with DataStream and DataSet API][]

*  在流处理方面，两种Planner都可以与 DataStream API 集成使用。
 *  只有旧planner可以与 DataSet API 结合。
 *  在批处理方面，Blink Planner不能同两种DataStream或DataSet集成。

### 2.4.2 隐式转换 ###

Scala Table API含有对DataSet、DataStream、 Table的隐式转换，DataStream API需要导入：

import org.apache.flink.table.api.bridge.scala._
    import org.apache.flink.api.scala._

### 2.4.3 通过 DataSet / DataStream创建临时视图 ###

视图的schema依赖于DataStream / DataSet注册的数据类型，详见[mapping of data types to table schema][]。

**注意： 通过DataStream或DataSet创建的视图只能注册为临时视图。**

// get TableEnvironment 
    // registration of a DataSet is equivalent
    val tableEnv: StreamTableEnvironment = ... // see "Create a TableEnvironment" section
    
    val stream: DataStream[(Long, String)] = ...
    
    // register the DataStream as View "myTable" with fields "f0", "f1"
    tableEnv.createTemporaryView("myTable", stream)
    
    // register the DataStream as View "myTable2" with fields "myLong", "myString"
    tableEnv.createTemporaryView("myTable2", stream, 'myLong, 'myString)

### 2.4.4 将 DataStream / DataSet 转换成Table ###

// get TableEnvironment
    // registration of a DataSet is equivalent
    val tableEnv = ... // see "Create a TableEnvironment" section
    
    val stream: DataStream[(Long, String)] = ...
    
    // convert the DataStream into a Table with default fields "_1", "_2"
    val table1: Table = tableEnv.fromDataStream(stream)
    
    // convert the DataStream into a Table with fields "myLong", "myString"
    val table2: Table = tableEnv.fromDataStream(stream, $"myLong", $"myString")
    
    val stream2: DataStream[(String, Int)] = ...
    // 如果没指定，则field从DataStream自动获取
    // DataStreamConversions#toTable,
    // DataSet时使用DataSetConversions#toTable
    val table = stream2.toTable(tEnv, 'name, 'amount)

### 2.4.5 将Table转换成DataStream / DataSet ###

#### 2.4.5.1 类型转换 ####

详细请参考[Mapping of Data Types to Table Schema][mapping of data types to table schema]

Table可转为DataStream/DataSet，这样可以使自定义DataStream/DataSet来使用Table&SQL 查询结果。

转变时，需要指定生成的DataStream/DataSet的数据类型，通常来说最方便的转换类型是`Row`，所有选项如下：  
![在这里插入图片描述][20200808234634676.png]

#### 2.4.5.2 Table转为DataStream ####

这里说的Table就是流式查询的结果，会随着新数据流入输入流中执行查询被不断地动态更新。因此，这种由动态查询转换成的DataStream需要对表的`UPDATE`进行编码。Table转为DataStream有两种模式：

*  Append Mode  
    仅当动态表只被`INSERT`操作改动被使用，是`append-only`的，而且旧的转换结果不会再被更新。
 *  Retract Mode  
    任何情况下都可以使用，使用 `boolean` 值对`INSERT`和`DELETE`操作的改动进行标记，`True`表示 INSERT, `False` 表示DELETE.。

示例代码：

// get TableEnvironment. 
    // registration of a DataSet is equivalent
    val tableEnv: StreamTableEnvironment = ... // see "Create a TableEnvironment" section
    
    // Table with two fields (String name, Integer age)
    val table: Table = ...
    
    // convert the Table into an append DataStream of Row
    val dsRow: DataStream[Row] = tableEnv.toAppendStream[Row](table)
    
    // convert the Table into an append DataStream of Tuple2[String, Int]
    val dsTuple: DataStream[(String, Int)] dsTuple = 
      tableEnv.toAppendStream[(String, Int)](table)
    
    // convert the Table into a retract DataStream of Row.
    // A retract stream of type X is a DataStream[(Boolean, X)]. 
    // The boolean field indicates the type of the change. 
    // True is INSERT, false is DELETE.
    val retractStream: DataStream[(Boolean, Row)] = tableEnv.toRetractStream[Row](table)

Table转为DataStream后，必须使用`StreamExecutionEnvironment.execute()`来运行该DataStream程序。

#### 2.4.5.3 Table转为DataSet ####

// get TableEnvironment 
    // registration of a DataSet is equivalent
    val tableEnv = BatchTableEnvironment.create(env)
    
    // Table with two fields (String name, Integer age)
    val table: Table = ...
    
    // convert the Table into a DataSet of Row
    val dsRow: DataSet[Row] = tableEnv.toDataSet[Row](table)
    
    // convert the Table into a DataSet of Tuple2[String, Int]
    val dsTuple: DataSet[(String, Int)] = tableEnv.toDataSet[(String, Int)](table)

Table转为DataSet后，必须使用`ExecutionEnvironment.execute()`来运行该DataSet程序。

## 2.5 创建表 ##

### 2.5.1 视图-虚拟表 ###

// get a TableEnvironment
    val tableEnv = ... // see "Create a TableEnvironment" section
    
    // table is the result of a simple projection query 
    val projTable: Table = tableEnv.from("X").select(...)
    
    // register the Table projTable as table "projectedTable"
    tableEnv.createTemporaryView("projectedTable", projTable)

内部其实就是封装了逻辑查询计划，但如果使用了Blink Planner，则在被多表引用时也只会被执行一次，结果可被多表共享。

### 2.5.2 创建connector表 ###

其实就是外表，描述了外部存储系统

tableEnvironment
      .connect(...)
      .withFormat(...)
      .withSchema(...)
      .inAppendMode()
      .createTemporaryTable("MyTable")

### 2.5.3 identifier ###

包括 catalog/database/table，可直接指定前两者，则使用可省略

// get a TableEnvironment
    val tEnv: TableEnvironment = ...;
    tEnv.useCatalog("custom_catalog")
    tEnv.useDatabase("custom_database")
    
    val table: Table = ...;
    
    // register the view named 'exampleView' in the catalog named 'custom_catalog'
    // in the database named 'custom_database' 
    tableEnv.createTemporaryView("exampleView", table)
    
    // register the view named 'exampleView' in the catalog named 'custom_catalog'
    // in the database named 'other_database' 
    tableEnv.createTemporaryView("other_database.exampleView", table)
    
    // register the view named 'example.View' in the catalog named 'custom_catalog'
    // in the database named 'custom_database' 
    tableEnv.createTemporaryView("`example.View`", table)
    
    // register the view named 'exampleView' in the catalog named 'other_catalog'
    // in the database named 'other_database' 
    tableEnv.createTemporaryView("other_catalog.other_database.exampleView", table)

## 2.6 查询 ##

// get a TableEnvironment
    val tableEnv = ... // see "Create a TableEnvironment" section
    
    // register Orders table
    
    // scan registered Orders table
    val orders = tableEnv.from("Orders")
    // compute revenue for all customers from France
    val revenue = orders
      .filter($"cCountry" === "FRANCE")
      .groupBy($"cID", $"cName")
      .select($"cID", $"cName", $"revenue".sum AS "revSum")
    
    // emit or convert Table
    // execute query

## 2.7 Emit输出表 ##

Table最终被输出到`TableSink`，他是一个通用接口，支持多种文件format、存储系统、MQ等。

*  流Table可写入AppendStreamTableSink / RetractStreamTableSink / UpsertStreamTableSink
 *  批Table只能写入BatchTableSink。

// get a TableEnvironment
    val tableEnv = ... // see "Create a TableEnvironment" section
    
    // create an output Table
    val schema = new Schema()
        .field("a", DataTypes.INT())
        .field("b", DataTypes.STRING())
        .field("c", DataTypes.LONG())
    
    tableEnv.connect(new FileSystem("/path/to/file"))
        .withFormat(new Csv().fieldDelimiter('|').deriveSchema())
        .withSchema(schema)
        .createTemporaryTable("CsvSinkTable")
    
    // compute a result Table using Table API operators and/or SQL queries
    val result: Table = ...
    
    // emit the result Table to the registered TableSink
    result.executeInsert("CsvSinkTable")

# 3 SQL API #

参考

*  [SQL API][]
 *  [依赖介绍][Link 2]

## 3.1 Table API & SQL 架构 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70 7]

## 3.2 执行原理 ##

*  [Flink 原理与实现：Table & SQL API][Flink _Table _ SQL API]  
    阿里-伍翀
 *  [基于Flink1.8 深入理解Flink Sql执行流程 + Flink Sql语法扩展][Flink1.8 _Flink Sql_ _ Flink Sql]
 *  [Flink | 使用Calcite做Sql语法解析][Flink _ _Calcite_Sql]

在1.11的Flink中，Table API不再被翻译为DataStream API，而是翻译为Transformation。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center]  
Flink 1.11 Sql执行流程如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 1]

1.  使用Calcite解析SQL，这个过程会用到Catalog（表名、列信息、类型信息、PK信息等，可用于逻辑计划的后续优化），翻译为Logic Plan
2.  将Logic Plan按一系列规则（如谓词下推、投影下推、Join重排等）翻译为Physical Plan
3.  通过Code Generation将Physical Plan转为Transformation DAG 图  
    这一阶段也有许多优化
4.  Transformation转为StreamGraph->JobGraph提交到Flink Cluster
5.  JobGraph生成带并行的信息的ExecutionGraph，最后生成虚拟的物理执行图运行

## 3.3 Blink和Old Planner ##

*  Blink是阿里贡献的
 *  Blink将batch job也作为特殊的流来处理（会被转为DataStream程序处理），所以不能转为DataSet。
 *  Blink不支持`BatchTableSource`，而是使用`StreamTableSource`
 *  Sink处理不同：
    
     *  Old Planner将Job中的多个Sink各自单独优化为一个DAG，每个DAG相互独立；
     *  Blink将多个Sink优化为一个DAG。
 *  Table&Sql翻译执行不同
    
     *  Blink将多个他们全部转为DataStream程序，不论输入是流还是批。一个查询在Flink内部被描述为逻辑查询计划，并被翻译为两个阶段：
        
         *  逻辑执行计划优化
         *  转换为一个DataStream程序  
            不同语句的Blink翻译时机转自官网：  
            ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70 8]
 *  我们实践中现在都用Blink了

## 3.4 示例 ##

### 3.4.1 基本结构 ###

批流统一的代码结构，以下是使用Blink时的scala示例：

// create a TableEnvironment for blink planner streaming
    val bsEnv = StreamExecutionEnvironment.getExecutionEnvironment
    val bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()
    // 每个表都绑定到一个特定的TableEnvironment，不能夸TableEnvironment操作表
    val tableEnv = StreamTableEnvironment.create(bsEnv, bsSettings)
    
    // 1. create a TableEnvironment for blink planner batch
    val bbSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build()
    val bbTableEnv = TableEnvironment.create(bbSettings)
    
    // 2. create Table 
    
    tableEnv.connect(...).createTemporaryTable("table1")
    // register an output Table
    tableEnv.connect(...).createTemporaryTable("outputTable")
    
    // create a Table from a Table API query
    val tapiResult = tableEnv.from("table1").select(...)
    // create a Table from a SQL query
    val sqlResult  = tableEnv.sqlQuery("SELECT ... FROM table1 ...")
    
    // emit a Table API result Table to a TableSink, same for SQL result
    val tableResult = tapiResult.executeInsert("outputTable")
    tableResult...

参见[Integration with DataStream and DataSet API][]可查看Table API和DataStream、DataSet整合、转换的例子。

### 3.4.2 创建connector表 ###

其实就是外表，描述了外部存储系统

tableEnvironment.executeSql("CREATE [TEMPORARY] TABLE MyTable (...) WITH (...)")

### 3.4.3 SQL ###

可参考[FlinkSQL查询][FlinkSQL]

查询指定表示例：

// get a TableEnvironment
    val tableEnv = ... // see "Create a TableEnvironment" section
    
    // register Orders table
    
    // compute revenue for all customers from France
    val revenue = tableEnv.sqlQuery("""
      |SELECT cID, cName, SUM(revenue) AS revSum
      |FROM Orders
      |WHERE cCountry = 'FRANCE'
      |GROUP BY cID, cName
      """.stripMargin)
    
    // emit or convert Table
    // execute query

更新指定表示例：

// get a TableEnvironment
    val tableEnv = ... // see "Create a TableEnvironment" section
    
    // register "Orders" table
    // register "RevenueFrance" output table
    
    // compute revenue for all customers from France and emit to "RevenueFrance"
    tableEnv.executeSql(""" |INSERT INTO RevenueFrance |SELECT cID, cName, SUM(revenue) AS revSum |FROM Orders |WHERE cCountry = 'FRANCE' |GROUP BY cID, cName """.stripMargin)

## 3.5 配置 ##

*  [Table-Configuration][]
 *  [通用配置][Link 3]

## 3.6 Catalog和Identifier标识符 ##

### 3.6.1 概述 ###

由`TableEnvironment`维护着一个由标识符（identifier）创建的表 catalog 的映射，标识符由`catalog`、`database`、`table`三部分组成。

用户可以指定默认的catalog和database

可以通过sql创建catalog

CREATE CATALOG catalog_name
      WITH (key1=val1, key2=val2, ...)

创建DATABASE

CREATE DATABASE [IF NOT EXISTS] [catalog_name.]db_name
      [COMMENT database_comment]
      WITH (key1=val1, key2=val2, ...)

Catalog可参考[catalog][Catalogs]

## 3.7 Table和View的创建 ##

参考

*  [CREATE Statements][]
 *  Flink Table描述了外部数据，比如文件、数据库、消息队列等。
    
        CREATE TABLE [catalog_name.][db_name.]table_name
          (
            { <column_definition> | <computed_column_definition> }[ , ...n]
            [ <watermark_definition> ]
            [ <table_constraint> ][ , ...n]
          )
          [COMMENT table_comment]
          [PARTITIONED BY (partition_column_name1, partition_column_name2, ...)]
          WITH (key1=val1, key2=val2, ...)
          [ LIKE source_table [( <like_options> )] ]
           
        <column_definition>:
          column_name column_type [ <column_constraint> ] [COMMENT column_comment]
          
        <column_constraint>:
          [CONSTRAINT constraint_name] PRIMARY KEY NOT ENFORCED
        
        <table_constraint>:
          [CONSTRAINT constraint_name] PRIMARY KEY (column_name, ...) NOT ENFORCED
        
        <computed_column_definition>:
          column_name AS computed_column_expression [COMMENT column_comment]
        
        <watermark_definition>:
          WATERMARK FOR rowtime_column_name AS watermark_strategy_expression
          
        <like_options>:
        {
           { INCLUDING | EXCLUDING } { ALL | CONSTRAINTS | PARTITIONS }
         | { INCLUDING | EXCLUDING | OVERWRITING } { GENERATED | OPTIONS | WATERMARKS } 
        }[, ...]
 *  Table也可以是虚表即`view`视图，可由对现存Table操作Table Api和SQL查询来创建。
    
        CREATE [TEMPORARY] VIEW [IF NOT EXISTS] [catalog_name.][db_name.]view_name
          [{ columnName [, columnName ]* }] [COMMENT view_comment]
          AS query_expression

## 3.8 临时表和永久表 ##

*  临时表的生命周期绑定到Flink Session  
    临时表存于内存，对其他Session不可见，也没有绑定catalog或数据库，但可以被创建到namespace中，但即使删除对应的namespace database也不会drop临时表！
 *  永久表一直存在直到被Drop，跨Flink Session和集群可见  
    永久表需要[catalog][Catalogs](如HiveCatalog)来存放表的元数据，对连接到该catalog的FlinkSession可见。
 *  Shadowing  
    如果注册和永久表同`identifier`的临时表，则临时表会掩盖永久表，此时永久表不可访问，所有对该标识符的表的访问都会路由到临时表。
    
    这个特性用来进行试验，一旦测试通过可以drop该临时表。

## 3.9 insert ##

参考

*  [INSERT Statement][]

## 3.10 Explain ##

参考

*  [EXPLAIN Statements][]
 *  [Explaining a Table][]
 *  sql方式

EXPLAIN PLAN FOR 
    	SELECT count, word FROM MyTable1 WHERE word LIKE 'F%' 
    	UNION ALL 
    	SELECT count, word FROM MyTable2;

还可以通过`TableEnvironment.explainSql()` 和 `TableEnvironment.executeSql()`执行

*  Table.explain()  
    得到表的查询计划
    
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        val tEnv = StreamTableEnvironment.create(env)
        
        val table1 = env.fromElements((1, "hello")).toTable(tEnv, $"count", $"word")
        val table2 = env.fromElements((1, "hello")).toTable(tEnv, $"count", $"word")
        val table = table1
          .where($"word".like("F%"))
          .unionAll(table2)
        println(table.explain())
 *  StatementSet.explain()  
    得到多个sink的查询计划
    
        val settings = EnvironmentSettings.newInstance.useBlinkPlanner.inStreamingMode.build
        val tEnv = TableEnvironment.create(settings)
        
        val schema = new Schema()
            .field("count", DataTypes.INT())
            .field("word", DataTypes.STRING())
        
        tEnv.connect(new FileSystem("/source/path1"))
            .withFormat(new Csv().deriveSchema())
            .withSchema(schema)
            .createTemporaryTable("MySource1")
        tEnv.connect(new FileSystem("/source/path2"))
            .withFormat(new Csv().deriveSchema())
            .withSchema(schema)
            .createTemporaryTable("MySource2")
        tEnv.connect(new FileSystem("/sink/path1"))
            .withFormat(new Csv().deriveSchema())
            .withSchema(schema)
            .createTemporaryTable("MySink1")
        tEnv.connect(new FileSystem("/sink/path2"))
            .withFormat(new Csv().deriveSchema())
            .withSchema(schema)
            .createTemporaryTable("MySink2")
            
        val stmtSet = tEnv.createStatementSet()
        
        val table1 = tEnv.from("MySource1").where($"word".like("F%"))
        stmtSet.addInsert("MySink1", table1)
        
        val table2 = table1.unionAll(tEnv.from("MySource2"))
        stmtSet.addInsert("MySink2", table2)
        
        val explanation = stmtSet.explain()
        println(explanation)

返回结果：

*  查询转换后得到的抽象语法树AST，他是尚未优化的逻辑查询计划
 *  优化后的逻辑查询计划
 *  物理查询计划

## 3.11 TableSink ##

TableSink指输出表：

*  可支持各种文件格式，如CSV, Apache Parquet, Apache Avro
 *  可支持各种存储系统，如JDBC、Apache HBase、Apache Cassandra、Elasticsearch
 *  可支持消息队列MQ，如 Apache Kafka、RabbitMQ

## 3.12 Blink对查询的优化 ##

Flink利用并扩展Calcite来执行查询优化，有基于规则和基于代价的优化：

*  子查询去相关性
 *  投影裁剪  
    投影解释：投影运算也是一个单目运算，它是从一个关系R中选取所需要的列组成一个新关系，就是`select a,b,c`这样的语句，让结果集仅包含指定列，这种操作称为投影查询。
 *  分区裁剪
 *  过滤器下推  
    否则会收集大量数据后再过滤
 *  子计划去重以避免重复计算
 *  特殊子查询重写，包括两部分：
    
     *  将`IN`和`EXISTS`转为`left semi-joins`
     *  将`NOT IN`和 `NOT EXISTS` 转换为 `left anti-join`
 *  可选的join重排序，通过`table.optimizer.join-reorder-enabled`开启

注意： 当前仅在子查询重写的结合条件下支持 IN / EXISTS / NOT IN / NOT EXISTS。

Flink优化器不仅基于查询计划，还基于从数据源获得的统计信息以及每个算子的细粒度成本（如 io，cpu，网络和内存）来做最佳决策。

高级用户还可以通过`CalciteConfig`提供自定义优化，可以通过调用 `TableEnvironment#getConfig#setPlannerConfig` 将该对象传递给 TableEnvironment。

## 3.13 [SQL Hints][] ##

### 3.13.1 概述 ###

Flink 1.11加入的新特性，可用来改变SQL的查询计划。

*  指定planner  
    Blink和默认的Planner都有各自最佳场景，所以可指定Planner
 *  追加元数据和统计数据  
    如`table index for scan`和`skew info of some shuffle keys`是动态变化的
 *  算子资源限制  
    多数情况给与算子默认资源配置（最小并行度或managed memory或指定的资源需求（GPC/SSD）等）。可通过Hints在Query级别控制资源。

### 3.13.2 Dynamic Table Options ###

可在每次Query时动态指定或覆写表的option。需要加上配置`set table.dynamic-table-options.enabled=true`来开启，默认关闭。

否则在查询该表时使用hints会报错：

Fail to run sql command: select * from xxx /*+ OPTIONS('scan.startup.mode'='earliest-offset') */
    org.apache.flink.table.api.ValidationException: OPTIONS hint is allowed only when table.dynamic-table-options.enabled is set to true

### 3.13.3 hints语法 ###

table_path /*+ OPTIONS(key=val [, key=val]*) */

*  key:  
    字符串类型的配置名
 *  val:  
    字符串类型的配置值

### 3.13.4 例子 ###

CREATE TABLE kafka_table1 (id BIGINT, name STRING, age INT) WITH (...);
    CREATE TABLE kafka_table2 (id BIGINT, name STRING, age INT) WITH (...);
    
    -- override table options in query source
    select id, name from kafka_table1 /*+ OPTIONS('scan.startup.mode'='earliest-offset') */;
    
    -- override table options in join
    select * from
        kafka_table1 /*+ OPTIONS('scan.startup.mode'='earliest-offset') */ t1
        join
        kafka_table2 /*+ OPTIONS('scan.startup.mode'='earliest-offset') */ t2
        on t1.id = t2.id;
    
    -- override table options for INSERT target table
    insert into kafka_table1 /*+ OPTIONS('sink.partitioner'='round-robin') */ select * from kafka_table2;

## 3.14 时态表 Temporal Table ##

### 3.14.1 概述 ###

注意和临时表（Temporary Table）区分！完全是不同概念。

时态表表示一个参数化的视图View，运行在一个变动中的Table上，返回指定时间点的该表的内容。

变动Table包括两类：

*  数据库表的changelog构成的changelog历史表  
    Flink可追踪变动，允许查询特定时刻表的内容。在Flink里，这种表由`Temporal Table Function`表示。
 *  物化变动构成的维度表  
    允许查询特定ProcessingTime时的表内容。在Flink里，这种表由`Temporal Table`表示。

### 3.14.2 目标 ###

#### 3.14.2.1 关联changing history table ####

Flink可追踪变动，允许查询特定时刻表的内容。在Flink里，这种表由`Temporal Table Function`表示。

可将`append-only`表的行解释为表的ChangeLog，返回指定时间点的该表的内容。将 append-only 的表解释为ChangeLog需要指定主键以及时间戳属性：

*  主键用来确定哪些行将被覆盖
 *  时间戳用来确定行有效的时间

时态表主要目标是简化此类查询，加速执行，减少State用量。

比如有下表

SELECT * FROM RatesHistory;
    
    rowtime currency   rate
    ======= ======== ======
    09:00   US Dollar   102
    09:00   Euro        114
    09:00   Yen           1
    10:45   Euro        116
    11:15   Euro        119
    11:49   Pounds      108

`RatesHistory`表表示其他币种兑换日元`Yen`的汇率，`append-only`。

如果要输出所有`10:58`时刻的汇率，SQL如下：

SELECT *
    FROM RatesHistory AS r
    WHERE r.rowtime = (
      SELECT MAX(rowtime)
      FROM RatesHistory AS r2
      WHERE r2.currency = r.currency
      AND r2.rowtime <= TIME '10:58');

结果：

rowtime currency   rate
    ======= ======== ======
    09:00   US Dollar   102
    09:00   Yen           1
    10:45   Euro        116

可以看到确实是`10:58`时刻的汇率，也就是该表的快照。

该例子中，`currency`即为主键，`rowtime`为时间戳属性列。

该表在Flink由`Temporal Table Function`表示。

#### 3.14.2.2 关联changing dimension table ####

允许查询特定ProcessingTime时的表内容。在Flink里，这种表由`Temporal Table`表示。

现有表`LatestRates`，内容是最新的汇率，也就是说他是物化的`RatesHistory`历史的物化终态表。

如果我们`10:58`（处理时间）的时候查询该表：

10:58> SELECT * FROM LatestRates;
    currency   rate
    ======== ======
    US Dollar   102
    Yen           1
    Euro        116

如果我们`12:00`（处理时间）的时候查询该表：

12:00> SELECT * FROM LatestRates;
    currency   rate
    ======== ======
    US Dollar   102
    Yen           1
    Euro        119
    Pounds      108

### 3.14.3 Temporal Table Function ###

使用时态表，要访问数据时，必须传递一个[时间属性][Link 4]，该属性用来确定将要返回的表的版本。

定义后，Temporal Table Function可用一个时间参数`timeAttribute`产生一些列行，该集合包含相对于给定时间属性的所有现有主键的行的最新版本。

比如在`RatesHistory`表上定义了一个Temporal Table Function：`Rates(timeAttribute)`，可返回给定时间属性列值时的Rates状态，则查询如下：

SELECT * FROM Rates('10:15');
    
    rowtime currency   rate
    ======= ======== ======
    09:00   US Dollar   102
    09:00   Euro        114
    09:00   Yen           1
    
    SELECT * FROM Rates('11:00');
    
    rowtime currency   rate
    ======= ======== ======
    09:00   US Dollar   102
    10:45   Euro        116
    09:00   Yen           1

**注意，其实目前Flink不支持直接传入常量到Temporal Table Function，目前只能应用在join中，上例只是为了让我们更好理解而已！**

对append-only table定义Temporal Table Function示例代码：

// Get the stream and table environments.
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val tEnv = StreamTableEnvironment.create(env)
    
    // Provide a static data set of the rates history table.
    val ratesHistoryData = new mutable.MutableList[(String, Long)]
    ratesHistoryData.+=(("US Dollar", 102L))
    ratesHistoryData.+=(("Euro", 114L))
    ratesHistoryData.+=(("Yen", 1L))
    ratesHistoryData.+=(("Euro", 116L))
    ratesHistoryData.+=(("Euro", 119L))
    
    // Create and register an example table using above data set.
    // In the real setup, you should replace this with your own table.
    val ratesHistory = env
      .fromCollection(ratesHistoryData)
      .toTable(tEnv, 'r_currency, 'r_rate, 'r_proctime.proctime)
    
    tEnv.createTemporaryView("RatesHistory", ratesHistory)
    
    // Create and register TemporalTableFunction.
    // Define "r_proctime" as the time attribute and "r_currency" as the primary key.
    // TemporalTableFunction内容
    val rates = ratesHistory.createTemporalTableFunction($"r_proctime", $"r_currency") 
    // 以Rates为名字注册该函数
    tEnv.registerFunction("Rates", rates)

### 3.14.4 Temporal Table ###

仅支持Blink planner。

为了访问时态表，当前必须使用 `LookupableTableSource` 定义一个 TableSource。可使用`FOR SYSTEM_TIME AS OF`语法来查询时态表。

查询时态表LatestRates：

SELECT * FROM LatestRates FOR SYSTEM_TIME AS OF TIME '10:15';
    
    currency   rate
    ======== ======
    US Dollar   102
    Euro        114
    Yen           1
    
    SELECT * FROM LatestRates FOR SYSTEM_TIME AS OF TIME '11:00';
    
    currency   rate
    ======== ======
    US Dollar   102
    Euro        116
    Yen           1

结果和上面使用`Temporal Table Function`的示例相同。

**注意，其实目前Flink不支持直接传入常量来查询Temporal Table，目前只能将时态表应用在join中，上例只是为了让我们更好理解而已！**

定义时态表：

// Get the stream and table environments.
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val settings = EnvironmentSettings.newInstance().build()
    val tEnv = StreamTableEnvironment.create(env, settings)
    // or val tEnv = TableEnvironment.create(settings)
    
    // Define an HBase table with DDL, then we can use it as a temporal table in sql
    // Column 'currency' is the rowKey in HBase table
    tEnv.executeSql(
        s"""
           |CREATE TABLE LatestRates (
           |    currency STRING,
           |    fam1 ROW<rate DOUBLE>
           |) WITH (
           |    'connector' = 'hbase-1.4',
           |    'table-name' = 'Rates',
           |    'zookeeper.quorum' = 'localhost:2181'
           |)
           |""".stripMargin)

LookupableTableSource定义请参考[这里][Link 5]

## 3.15 Join ##

### 3.15.1 概述 ###

动态表的join概念比普通静态表要难理解一些。

关于FlinkSql支持的Join可参照[这里][Link 6]

以下开始讲不同Join场景。

### 3.15.2 Regular Join ###

指两个流的join。

这种场景下，两个表中的新数据、旧数据改动都会造成新的join计算。比如左表来了条新数据，则会和右表的旧的、未来的数据全部计算join。

这个join语法适用于存在任意修改类型（insert, update, delete）的输入表。

但这样Join需要被join的两个表的输入永久保存在Flink的State里。因此，使用的资源量也会随着输入表的增长而无限增长。所以需要配置数据状态驻留期限。参阅[Query Configuration][]

*  Inner Equi-join  
    目前只支持等值连接，即join条件至少包含一个`=`的维持，不支持任何 cross join 和 theta join。
    
    还要注意,Join的顺序未优化，目前会按照sql中`from`后表定义的顺序依次执行join，需要确保该顺序不会导致cross join（笛卡儿积），否则会导致失败。
    
    最后需要注意的是需要配置State过期时间，否则会无限增长。参阅[Query Configuration][]
    
    正例如下：
    
        SELECT * FROM Orders
        INNER JOIN Product
        ON Orders.productId = Product.id
 *  Outer Equi-join  
    目前只支持等值连接，还要注意,Join的顺序未优化，也需要配置State过期时间，同`Inner Equi-join`
    
    正例如下：
    
        SELECT *
        FROM Orders LEFT JOIN Product ON Orders.productId = Product.id
        
        SELECT *
        FROM Orders RIGHT JOIN Product ON Orders.productId = Product.id
        
        SELECT *
        FROM Orders FULL OUTER JOIN Product ON Orders.productId = Product.id

### 3.15.3 Interval Join(时间区间关联) ###

是`Regular Join`的子集，可以使用流的方式进行处理。

不同的是，`Interval Join`只支持带有时间属性的`append-only`表，不支持删改。

而且，由于时间属性是准单调递增的，因而Flink可以从State移除旧的值，而不会影响结果正确性。

执行时，Interval join需要至少一个等值连接谓词和一个限制了两表的时间的 join 条件。如使用两个适当的Range谓词（<, <=, >=, >），一个 `BETWEEN` 谓词或单个比较两个输入表中相同事件类型(process/event time)的时间属性的等值谓词。

正例如下

ltime = rtime
    ltime >= rtime AND ltime < rtime + INTERVAL '10' MINUTE
    ltime BETWEEN rtime - INTERVAL '10' SECOND AND rtime + INTERVAL '5' SECOND

将所有收到订单后4小时内发货的订单表和他们对应的发货单进行join的例子：

SELECT *
    FROM Orders o, Shipments s
    WHERE o.id = s.orderId AND
          o.ordertime BETWEEN s.shiptime - INTERVAL '4' HOUR AND s.shiptime

### 3.15.4 时态表函数 Join ###

#### 3.15.4.1 概述 ####

这是将一个`append-only table`作为左表和时态表作为右表进行join。

以下例子展示了一个join的例子。

*  Orders  
    append-only table

SELECT * FROM Orders;
    
    rowtime amount currency
    ======= ====== =========
    10:15        2 Euro
    10:30        1 US Dollar
    10:32       50 Yen
    10:52        3 Euro
    11:04        5 US Dollar

*  RatesHistory  
    时态表，具体是个append-only table类型的各币种对日元(Yen)汇率变动。该表在Flink由`Temporal Table Function`表示。

如果要将Orders表的金额转为日元，一般会这么做

SELECT
      SUM(o.amount * r.rate) AS amount
    FROM Orders AS o,
      RatesHistory AS r
    WHERE r.currency = o.currency
    AND r.rowtime = (
      SELECT MAX(rowtime)
      FROM RatesHistory AS r2
      WHERE r2.currency = o.currency
      AND r2.rowtime <= o.rowtime);

而如果使用 `Temporal Table Function` ，注册一个在时态表`RatesHistory`上的函数`Rates`，可大大简化：

SELECT
      o.amount * r.rate AS amount
    FROM
      Orders AS o,
      LATERAL TABLE (Rates(o.rowtime)) AS r
    WHERE r.currency = o.currency

每条probe左表的数据会和build右表的对应时间属性版本的所有数据行做jon。也就是说，上例中`Orders`表的每条记录会和右表`Rates`在对应的`o.rowtime`的版本的数据做join。

右表需要主键以修改，这里是`currency`列。

如果使用ProcessingTime来查询，则新append的order总是会join最新版本的`Rates`数据。

可以发现，时态表Join和常规Join最大不同就是，右表（时态表）有新数据也不会影响之前的join结果。这可限制Flink在State中保存元素的数量。也就是说，随着时间推移，之前的不再需要的版本的数据会从State中移除。

#### 3.15.4.2 使用 ####

注意，目前时态表join时的状态过期尚未实现（Fink1.11），可能导致查询使用的状态无限增长。

关于ProcessingTime和EventTime时的时态表join可参考[这里][Link 7]

*  ProcessingTime  
    此时不能传递过去的时间属性给temporal table function，而只能总是当前时间。所以这种场景下每次调用temporal table function总是返回最新的已知版本的表数据。
    
    注意，此时仅build表的最新版本（相对于定义的主键）的数据会被保存在状态中。
    
    build表的更新不会影响先前发出的join结果，因为只会让左表和build表最新版本数据join。
    
    可以将这种join想象为`HashMap<K, V>`，存有所有build表的行。如果新数据有着和旧数据重复key，此时直接覆盖该缓存即可。此时每个probe左表总是使用最新的HashMap。
 *  EventTime  
    此时可传递过去的时间给temporal table function，进行join。
    
     *  EventTime  
        此时可传递过去的时间给temporal table function，进行join。
    
    此场景下，会将自从上次水位依赖所有版本的行保存到State中。水位的作用是推进join算子进程，丢弃不需要的版本的build table数据。
    
    举个例子，比如拥有`12:30:00`的事件时间戳的行append到probe表，此时就会和时态表`12:30:00`版本的数据（即时间戳小于等于该时间的数据）进行join，并会不断更新主键，并根据主键应用更新直到该时间点。

### 3.15.5 时态表 Join ###

#### 3.15.4.1 概述 ####

这是将一个包含任意操作的表作为probe左表和时态表(必须由[LookupableTableSource][]支撑)作为build右表进行join。

比如我们有一个不断物化变化的当前汇率表`LatestRates`作为时态表：

10:15> SELECT * FROM LatestRates;
    
    currency   rate
    ======== ======
    US Dollar   102
    Euro        114
    Yen           1
    
    10:30> SELECT * FROM LatestRates;
    
    currency   rate
    ======== ======
    US Dollar   102
    Euro        114
    Yen           1
    
    
    10:52> SELECT * FROM LatestRates;
    
    currency   rate
    ======== ======
    US Dollar   102
    Euro        116     <==== changed from 114 to 116
    Yen           1

另有一个`append-only`表Orders：

SELECT * FROM Orders;
    
    amount currency
    ====== =========
         2 Euro             <== arrived at time 10:15
         1 US Dollar        <== arrived at time 10:30
         2 Euro             <== arrived at time 10:52

以下例子展示了普通标Orders和时态表LatestRates join以计算订单金额转换为当时汇率对应的日元的例子。

期待的结果如下：

amount currency     rate   amout*rate
    ====== ========= ======= ============
         2 Euro          114          228    <== arrived at time 10:15
         1 US Dollar     102          102    <== arrived at time 10:30
         2 Euro          116          232    <== arrived at time 10:52

joinSQL:

SELECT
      o.amout, o.currency, r.rate, o.amount * r.rate
    FROM
      Orders AS o
      JOIN LatestRates FOR SYSTEM_TIME AS OF o.proctime AS r
      ON r.currency = o.currency

每条probe左表的数据会和build右表的当前版本的数据行做join。也就是说，上例中`Orders`表的每条新记录会和右表`LatestRates`最新版本的数据做join，因为这里使用的是`proctime`。

可以发现，时态表Join和常规Join最大不同就是，右表（时态表）有新数据也不会影响之前的join结果。同时，时态表join算子十分轻量级，不需要保存任何State！

#### 3.15.5.2 使用 ####

时态表join语法：

SELECT [column_list]
    FROM table1 [AS <alias1>]
    [LEFT] JOIN table2 FOR SYSTEM_TIME AS OF table1.proctime [AS <alias2>]
    ON table1.column-name1 = table2.column-name1

*  目前仅支持`Blink planner`
 *  目前仅支持SQL API，不支持Table API
 *  目前仅支持ProcessingTime 时态表join，不支持EventTime
 *  目前仅支持`INNER JOIN`和`LEFT JOIN`
 *  `FOR SYSTEM_TIME AS OF`必须跟在时态表之后
 *  proctime是probe表table1的处理时间属性，含义是join时，会获取时态表在处理时间的快照进行join

实例：

SELECT
      SUM(o_amount * r_rate) AS amount
    FROM
      Orders
      JOIN LatestRates FOR SYSTEM_TIME AS OF o_proctime
      ON r_currency = o_currency

### 3.15.6 小结 ###

*  Temporal Tables 是跟随时间变化而变化的表。
 *  Temporal Table Function 提供访问 Temporal Tables 在某一时间点的状态的能力
 *  Join Temporal Table Function 的语法与 Join Table Function 一致
 *  目前仅支持在 Temporal Tables 上的 inner join

temporal table function join 对比 temporal table join：

*  相同：
    
     *  目标相同
 *  不同：
    
     *  SQL语法不同  
        前者使用join UDTF，后者使用标准的时态表语法(SQL:2011)
     *  State保存不同  
        前者将join的两个流保存在State，后者仅接受输入流，并在外部数据库查找对应key的记录
     *  前者一般用来和changelog stream做join，后者一般和外部表（维表）做join

### 3.15.7 Join with Table Function (UDTF) ###

将表与另一个表的函数执行的结果进行 join 操作。左表（outer）中的每一行将会与右表的函数执行所产生的所有结果中相关联的行进行 join 。

当然，要先注册[UDTF][]。

*  Inner Join  
    此时如果函数调用结果为空，则对应的左表行会被丢弃

SELECT users, tag
    FROM Orders, LATERAL TABLE(unnest_udtf(tags)) t AS tag

*  Left Outer Join  
    此时如果函数调用结果为空，则对应的左表行会被保留，使用`null`填充

SELECT users, tag
    FROM Orders LEFT JOIN LATERAL TABLE(unnest_udtf(tags)) t AS tag ON TRUE

# 4 Connector的前置概念 #

## 4.1 概述 ##

可参考:

*  [Connect to External Systems][]

DDL不行，我们可以用[Connect to External Systems][Connect to External Systems 1]，直接读写外部数据源流批数据：

用来定义外部数据源连接。不是所有都支持流/批，支持批的Connector支持的Update Mode也不尽相同。

支持的输出：

*  可支持各种文件格式，如CSV, Apache Parquet, Apache Avro  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70 9]
 *  可支持各种存储系统，如JDBC、Apache HBase、Apache Cassandra、Elasticsearch、[Hive][Hive 1]  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70 10]
 *  可支持消息队列MQ，如 Apache Kafka、RabbitMQ

注意，从1.11.0开始相关API有巨大变化，要用老的请查看[legacy documentation][]

*  Table Schema  
    定义表的schema，描述了怎么将Table Source的数据格式映射到Table API的schema，以及Table映射到Sink的方式。可暴露给SQL查询。
 *  支持Time属性  
    可以使用一个或多个字段来提取或插入时间属性到Table Schema。

Flink连接外部系统可通过以下两种方式指定：

*  使用 Table & SQL API，搭配`org.apache.flink.table.descriptors`下的内容
 *  通过SQL客户端的YAML配置文件声明

一个Table & SQL API中连接外部数据源语句基本结构：

tableEnvironment
    // 定义连接外部数据源
      .connect(...)
      // 定义解析外部数据源中数据格式
      .withFormat(...)
      // 定义流式数据表的schema
      .withSchema(...)
      // 定义输出表的更新模式（update modes）
      .inAppendMode()
      // 注册Source表到flink
      .registerTableSource("MyTable")
      // 注册Sink表到flink
      .registerTableSink
      // 使用相同名字注册Source和Sink表
      .registerTableSourceAndSink

一个从Kafka中读Avro格式存储的数据的例子：

tableEnvironment
    // 定义连接外部Kafka数据源的配置
    .connect(
      new Kafka()
        .version("0.10")
        .topic("test-input")
        .startFromEarliest()
        .property("zookeeper.connect", "localhost:2181")
        .property("bootstrap.servers", "localhost:9092")
    )
    
    // 定义解析外部kafka数据源中数据格式
    .withFormat(
      new Avro()
        .avroSchema(
          "{" +
          " \"namespace\": \"org.myorganization\"," +
          " \"type\": \"record\"," +
          " \"name\": \"UserMessage\"," +
          " \"fields\": [" +
          " {\"name\": \"timestamp\", \"type\": \"string\"}," +
          " {\"name\": \"user\", \"type\": \"long\"}," +
          " {\"name\": \"message\", \"type\": [\"string\", \"null\"]}" +
          " ]" +
          "}"
        )
    )
    
    // 定义流式数据表的schema
    .withSchema(
      new Schema()
        .field("rowtime", Types.SQL_TIMESTAMP)
          .rowtime(new Rowtime()
            .timestampsFromField("timestamp")
            .watermarksPeriodicBounded(60000)
          )
        .field("user", Types.LONG)
        .field("message", Types.STRING)
    )
    
    // 定义输出表的更新方式为append
    .inAppendMode()
    
    // 同时注册source和sink
    .registerTableSource("MyUserSourceTable");
    //registerTableSink可参考《深入理解flink》243页
    .registerTableSink("MyUserSinkTable");

配置的连接属性会被转换为标准化的、基于String的key-value键值对。会基于Java SPI机制搜索唯一匹配的Table Factory来创建Table Source、Table Sink以及相应的format。

## 4.2 示例 ##

我们需要使用FlinkSql定义表名、表的Schema以及表的属性，以便连接外部系统。

以下注册Kafka数据源+读取json数据的例子：

CREATE TABLE MyUserTable (
      -- 定义流式数据表的schema，包括字段、水位等，其实就是flink内部和外部系统的映射关系
      `user` BIGINT,
      message STRING,
      ts TIMESTAMP,
      proctime AS PROCTIME(), -- use computed column to define proctime attribute
      WATERMARK FOR ts AS ts - INTERVAL '5' SECOND  -- use WATERMARK statement to define rowtime attribute
    ) WITH (
      -- declare the external system to connect to
      'connector' = 'kafka',
      'topic' = 'topic_name',
      'scan.startup.mode' = 'earliest-offset',
      'properties.bootstrap.servers' = 'localhost:9092',
      'format' = 'json'   -- declare a format for this system
    )

由[dynamic-table-factories][]（基于[java SPI][]查找）来根据sql创建TableSource、TableSink以及对应的`format`，。

## 4.3 主键 ##

1.11.0以前靠的是`group by`后的关键字进行推断，1.11.0以后可以指定。

主键约束表名一个或若干列是一个表内的唯一一行，且不包含null值。

在SinkTable内使用主键，一般被用来做`upsert`操作。

由用户来确保查询强制执行键完整性。

CREATE TABLE MyTable (
      MyField1 INT,
      MyField2 STRING,
      MyField3 BOOLEAN,
      PRIMARY KEY (MyField1, MyField2) NOT ENFORCED  -- defines a primary key on columns
    ) WITH (
      ...
    )

## 4.4 时间属性 ##

### 4.4.1 Processing time ###

即执行相应算子所在机器系统时间(`wall-clock time`墙上时间)，最简单的时间属性，但有不确定性。

该属性不需要时间戳提取器（timestamp extract）和水位生成器。

*  DDL
    
        CREATE TABLE user_actions (
          user_name STRING,
          data STRING,
          user_action_time AS PROCTIME() -- 定义user_action_time列，作为processingTime属性列
        ) WITH (
          ...
        );
        
        SELECT TUMBLE_START(user_action_time, INTERVAL '10' MINUTE), COUNT(DISTINCT user_name)
        FROM user_actions
        GROUP BY TUMBLE(user_action_time, INTERVAL '10' MINUTE);
 *  DataStream-to-Table转换时定义  
    必须在schema最后声明processingTime属性列
    
        val stream: DataStream[(String, String)] = ...
        
        // 声明一个额外的逻辑字段$"user_action_time"作为processingTime属性列
        // 所以，必须在schema最后声明processingTime属性列
        val table = tEnv.fromDataStream(stream, $"UserActionTimestamp", $"user_name", $"data", $"user_action_time".proctime)
        
        val windowedTable = table.window(Tumble over 10.minutes on $"user_action_time" as "userActionWindow")
 *  TableSource

// define a table source with a processing attribute
    // 注意TableSource需要实现DefinedProctimeAttribute接口
    class UserActionSource extends StreamTableSource[Row] with DefinedProctimeAttribute { 
    	
    	override def getReturnType = { 
    		val names = Array[String]("user_name" , "data")
    		val types = Array[TypeInformation[_]](Types.STRING, Types.STRING)
    		Types.ROW(names, types)
    	}
    
    	override def getDataStream(execEnv: StreamExecutionEnvironment): DataStream[Row] = { 
    		// create stream
    		val stream = ...
    		stream
    	}
    	// 定义逻辑processingTime属性列
    	override def getProctimeAttribute = { 
    		// field with this name will be appended as a third field
    		"user_action_time"
    	}
    }
    
    // register table source
    tEnv.registerTableSource("user_actions", new UserActionSource)
    
    val windowedTable = tEnv
    	.from("user_actions")
    	.window(Tumble over 10.minutes on $"user_action_time" as "userActionWindow")

### 4.4.2 Event time ###

指用户附加给每行数据的时间戳，Flink基于该时间戳作为时间基准进行处理。

特点是可以处理一定范围内的乱序事件和迟到时间（一致性保证），可保证从同一个存储层内反复读取同样一批数据时结果是可重放的。

将Event time设为时间属性时，Flink需要从数据记录中提取时间戳以及生成水位，以便处理乱序和迟到事件。

*  DDL
    
        CREATE TABLE user_actions (
          user_name STRING,
          data STRING,
          user_action_time TIMESTAMP(3),
          -- declare user_action_time as event time attribute and use 5 seconds delayed watermark strategy
          WATERMARK FOR user_action_time AS user_action_time - INTERVAL '5' SECOND
        ) WITH (
          ...
        );
        
        SELECT TUMBLE_START(user_action_time, INTERVAL '10' MINUTE), COUNT(DISTINCT user_name)
        FROM user_actions
        GROUP BY TUMBLE(user_action_time, INTERVAL '10' MINUTE);
 *  DataStream-to-Table转换时定义  
    待转换的DataStream必须已经定义时间戳和水位。
    
    有两种方法定义DataStream中的EventTime 时间戳：
    
     *  x.rowtime时x字段不存在  
        在 schema 的结尾追加一个新的字段
     *  x.rowtime时x字段存在  
        替换已经存在的x字段
    
        // Option 1:
        
        // extract timestamp and assign watermarks based on knowledge of the stream
        val stream: DataStream[(String, String)] = inputStream.assignTimestampsAndWatermarks(...)
        
        // declare an additional logical field as an event time attribute
        val table = tEnv.fromDataStream(stream, $"user_name", $"data", $"user_action_time".rowtime)
        
        
        // Option 2:
        
        // extract timestamp from first field, and assign watermarks based on knowledge of the stream
        val stream: DataStream[(Long, String, String)] = inputStream.assignTimestampsAndWatermarks(...)
        
        // the first field has been used for timestamp extraction, and is no longer necessary
        // replace first field with a logical event time attribute
        val table = tEnv.fromDataStream(stream, $"user_action_time".rowtime, $"user_name", $"data")
        
        // 使用:
        val windowedTable = table.window(Tumble over 10.minutes on $"user_action_time" as "userActionWindow")
 *  TableSource

// 定义一个有EventTime属性的 TableSource
    // 必须实现DefinedRowTimeAttributes接口
    class UserActionSource extends StreamTableSource[Row] with DefinedRowtimeAttributes { 
    
    	override def getReturnType = { 
    		val names = Array[String]("user_name" , "data", "user_action_time")
    		val types = Array[TypeInformation[_]](Types.STRING, Types.STRING, Types.LONG)
    		Types.ROW(names, types)
    	}
    	
    	// 需要确保此方法返回的DataStream 已经定义好了时间属性。
    	// 1.定义StreamRecordTimestamp，从DataStream提取时间戳
    	// 2.定义PreserveWatermarks，以保留DataStream的水位
    	// 否则，只有TableSource的 rowtime属性有效
    	override def getDataStream(execEnv: StreamExecutionEnvironment): DataStream[Row] = { 
    		// 构造 DataStream
    		// ...
    		// 基于 "user_action_time" 定义 watermark
    		val stream = inputStream.assignTimestampsAndWatermarks(...)
    		stream
    	}
    	// 本方法用来返回RowtimeAttributeDescriptor列表
    	// 内部包含了EventTime属性字段的名字、计算方法、水位生成策略等信息
    	override def getRowtimeAttributeDescriptors: util.List[RowtimeAttributeDescriptor] = { 
    		// 标记 "user_action_time" 字段是事件时间字段
    		// 计算方法是使用 "user_action_time" 字段构建
    		// 水位生成策略为AscendingTimestamps周期性递增式水位
    		val rowtimeAttrDescr = new RowtimeAttributeDescriptor(
    			"user_action_time",
    			new ExistingField("user_action_time"),
    			new AscendingTimestamps)
    		val listRowtimeAttrDescr = Collections.singletonList(rowtimeAttrDescr)
    		listRowtimeAttrDescr
    	}
    }
    
    // register the table source
    tEnv.registerTableSource("user_actions", new UserActionSource)
    
    val windowedTable = tEnv
    	.from("user_actions")
    	.window(Tumble over 10.minutes on $"user_action_time" as "userActionWindow")

### 4.4.3 Ingestion time ###

数据到达Flink Source的时间，内部处理方式类似Event time

注意：只要时间属性没有被修改过，而是仅仅被传递，那就一直保持为合法时间属性，可以和普通时间戳一样被访问计算，但只要被计算就会被物化、称为常规时间戳，就不能再作为Flink时间属性了，也不能在水位系统中起作用了，更不能作用到时间相关计算！

### 4.4.4 时间属性使用 ###

*  时间属性种类指定

val env = StreamExecutionEnvironment.getExecutionEnvironment
    
    env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime) // default
    
    // 可选
    // env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime)
    // env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

*  时间属性字段指定  
    如Window等基于时间的算子，就需要使用时间属性，所以应该在建表时指定时间属性字段。指定时机如下：
    
     *  DDL `CREATE TABLE`
     *  DataStream
     *  TableSource
 *  时间属性字段传递  
    时间属性可被传递，只要没有修改，就可以在一个表简单传递到另一个表后继续使用。
 *  时间属性字段物化  
    当时间属性字段被用在计算中时，就会被物化为普通时间戳，该时间戳不能配合Flink时间和水位使用了，也就无法用在基于时间的算子中

## 4.5 [Connector原理][Connector] ##

### 4.5.1 概述 ###

`DynamicSource` 和 `DynamicSink`可被用来从外部系统读或写入数据到外部系统，他们也可以被称为`Connector`。

下图展示了在翻译过程中，从一个stage到下一个stage时，Object如何被转为其他Object：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70 11]

### 4.5.2 Metadata ###

动态表的元数据（包括DDL和由Catalog提供的）由`CatalogTable`实例表示。

对应上图的Catalog->CatalogTable。

### 4.5.3 Planning ###

#### 4.5.3.1 概述 ####

利用CatalogTable来生成具体Connector相关的逻辑执行计划。

当开始为Table生成执行计划和优化，CatalogTable被解析为`DynamicTableSource`（用在`SELECT`语句中读取）和`DynamicTableSink`（用在`INSERT INTO`语句中写入）。

解析过程中`DynamicTableSourceFactory`（生成`DynamicTableSource`具体实现类，如`JdbcDynamicTableSource`）和`DynamicTableSinkFactory`（生成`DynamicTableSink`具体实现类，如`JdbcDynamicTableSink`）接口的具体实现类就会根据指定的Connector逻辑来翻译CatalogTable元数据。Dynamic Factory被用来根据catalog和session信息中为外部存储系统配置动态表Connector。这些工厂类的目的是验证DDL选项（如图中的`'port' = '5022'`）、配置编解码格式等，最后创建参数化的Table Connector实例。

#### 4.5.3.2 SPI ####

这两个工厂类接口的实现类查找和初始化是通过java SPI机制实现，也就是说DDL中的`'connector' = 'custom'`必须能找到对应的工厂类。SPI将检查唯一匹配的工厂类，该工厂类由工厂标识符和请求的父类名字（例如`DynamicTableSourceFactory`）唯一标识。

比如可以看看`flink-connector-jdbc`相关SPI定义文件:

*  META-INF/services/org.apache.flink.table.factories.Factory  
    org.apache.flink.connector.jdbc.table.JdbcDynamicTableFactory
 *  META-INF/services/org.apache.flink.table.factories.TableFactory  
    org.apache.flink.connector.jdbc.catalog.factory.JdbcCatalogFactory  
    org.apache.flink.connector.jdbc.table.JdbcTableSourceSinkFactory

更多关于SPI内容可参考:

*  [Java SPI 机制在 Flink SQL 中的应用][Java SPI _ Flink SQL]
 *  [Flink1.10基于工厂模式的任务提交与SPI机制][Flink1.10_SPI]

如果需要，Catalog也可以绕过这个工厂SPI发现机制，那就需要 `org.apache.flink.table.catalog.Catalog#getFactory`返回一个实现了所请求的父类的实现类的实例。

FlinkSql Planner会使用Source和Sink实例来执行指定Connector相关的相互通信，直到发现最佳逻辑执行计划。

依赖于可选的接口（如SupportsProjectionPushDown或SupportsOverwrite等），planner也许会将修改应用到生成的实例上，并修改运行时实现。

#### 4.5.3.3 DynamicTableSource ####

套路为`DynamicTableSourceFactory -> DynamicTableSource -> ScanRuntimeProvider`，工厂类负责翻译DDL option，Source负责创建运行时逻辑。

当读取动态表时，读到的数据可能被视为以下两个：

*  ScanTableSource  
    ChangeLog，Source在batch场景发送有界的、insert-only流；Streaming场景发送无界的、insert-only流；CDC(Change Data Capture)场景发送无界/有界的、可包含增删改的记录的流。
    
    从外部数据中扫描整张表的所有行。可包含增删改的行，所以可被用来读取changelog。
    
    `getChangelogMode`返回`ChangelogMode`，表示运行时可能返回的Changelog类型集合。
    
    如果Source有一些额外的能力，可以实现`org.apache.flink.table.connector.source.abilities`包内的接口，如`SupportsProjectionPushDown`投影下推。
    
    ScanTableSource生产的记录必须是`org.apache.flink.table.data.RowData`，用来在Flink内部传递。
 *  LookupTableSource  
    一个持续变化或非常大的外部表，这个表的内容一般不会被整个读取，而是在需要时单独读取。
    
    在运行时，通过一个或多个Key来从外部系统中查找数据行。不需要读取整张表，可以懒惰地获取单独的数据。
    
    注意，LookupTableSource目前仅支持`insert-only`变更；不支持abilities包内的接口。
    
    LookupTableSource的运行时实现是`TableFunction`或`AsyncTableFunction`。 在运行期间，将使用给定查找key的值来调用该函数。

Class可以同时实现以上两个接口，由Flink Planner来根据具体的查询内容来决定使用的方法。

#### 4.5.3.4 DynamicTableSink ####

当写入动态表时，数据可被视为ChangeLog。`getChangelogMode`方法返回`ChangelogMode`，表示运行时可能返回的Changelog类型集合。

*  Sink在batch场景只能接收insert-only行，输出有界流；
 *  Sink在Streaming场景只能接收insert-only行，输出无界流；
 *  Sink在CDC(Change Data Capture)场景可接收增删改的行，输出无界/有界流。

可实现额外功能，详见`org.apache.flink.table.connector.sink.abilities`包，如`SupportsOverwrite`。

与`DynamicTableSource`输出`org.apache.flink.table.data.RowData`对应的，DynamicTableSink必须消费该类结构的数据。

#### 4.5.3.5 Encoding / Decoding Format ####

一些Table Connector接收不同的format以编解码key和value。Format实现层次类似`DynamicTableSourceFactory -> DynamicTableSource -> ScanRuntimeProvider`，发现机制也是用的SPI。

### 4.5.4 Runtime ###

逻辑执行计划 -> 物理执行计划。

一旦逻辑执行计划生成，Planner会从具体的TableConnector中生成运行时实现逻辑，这些运行逻辑是Flink Connector的接口如`InputFormat`、`SourceFunction`等的实现类实现的。

这些接口按其他层级的抽象，分组为`ScanRuntimeProvider`、 `LookupRuntimeProvider`、 `SinkRuntimeProvider`的子类。

比如`OutputFormatProvider`(提供`org.apache.flink.api.common.io.OutputFormat`)和`SinkFunctionProvider` (提供 `org.apache.flink.streaming.api.functions.sink.SinkFunction`)都是`SinkRuntimeProvider`的子类。

## 4.6 Table Schema ##

### 4.6.1 概述 ###

Table Schema定义表的每个列的名字和类型，类似于SQL `create table`语句那样，用来暴露给SQL查询。此外，还可以指定如何将列与表数据编码schema的字段进行映射。当输入列无序时，Tabel Schema可清晰地定义列名、顺序和来源。Table Schema会和Table Format匹配来在Table数据输入和输出的过程中完成Schema转换。

此外， Table Schema还可指定Time属性提取器。

### 4.6.2 例子 ###

简单例子：

.withSchema(
      new Schema()
      	// 必填。和数据源中列顺序一致来指定Flink数据表的列
        .field("MyField1", Types.INT)
        .field("MyField2", Types.STRING)
        .field("MyField3", Types.BOOLEAN)
    )

复杂例子：

.withSchema(
      new Schema()
        .field("MyField1", Types.SQL_TIMESTAMP)
        	// 可选的，指定该列为processing-time 
          .proctime()     
        .field("MyField2", Types.SQL_TIMESTAMP)
        	// 可选的，指定该列为event-time(rowtime)
          .rowtime(...)   
        .field("MyField3", Types.BOOLEAN)
        	// 可选的，指定该列的原始来源列为mf3
          .from("mf3")

### 4.6.3 Rowtime ###

上述的`.rowtime(...)`本小节详细说下。

rowtime在flink里用来处理事件时间event-time。

采用Rowtime时，总是需要设置timestamp提取策略和watermark策略。

timestamp提取为rowtime例子如下：

.rowtime(
      new Rowtime()
      	// 转换input中的某个LONG或SQL_TIMESTAMP类型的列为rowtime
        .timestampsFromField("ts_field")  
    )
    
    .rowtime(
      new Rowtime()
      	// 使用input数据中的timestamp属性来转为rowtime，需要数据源支持，如Kafka 0.10+版本
        .timestampsFromSource()
    )
    
    .rowtime(
      new Rowtime()
      	// 为rowtime设置一个自定义的timestamp转换器
      	// ，该转换器必须实现自 org.apache.flink.table.sources.tsextractors.TimestampExtractor
        .timestampsFromExtractor(...)
    )

水位策略例子：

.rowtime(
      new Rowtime()
      	// 为升序的rowtime设置watermark。
      	// 发出截止目前观察到的最大timestamp-1的watermark。
      	// timestamp等于最大timestamp的行不算迟到。
        .watermarksPeriodicAscending()
    )
    
    .rowtime(
      new Rowtime()
      	// 为rowtime设置一个内嵌的watermark，该rowtime属性在有限的时间间隔内是乱序的
      	// 发出截止目前观察到的最大timestamp减去指定延迟（毫秒）的watermark。
        .watermarksPeriodicBounded(2000)
    )
    
    
    .rowtime(
      new Rowtime()
      	// 设置一个内置水印策略，该策略指示应从DataStream API中保留水印，从而保留数据源中分配的水印。
        .watermarksFromSource()
    )

## 4.7 Table Formats ##

### 4.7.1 概述 ###

一些外部数据系统支持不同的Table Formats，比如kafka或文件就支持其内存储的表的行使用CSV、JSON、Avro进行编码，所以需要指定Table Format来阐明外部数据源解析方式。

### 4.7.2 JSON Table Format ###

JSON格式允许读取和写入与给定的`format schema`相对应的JSON数据。format schema可用Flink type（SQl-like，映射到对应的SQL数据类型）、 JSON schema（适合复杂的嵌套数据结构）或目标表的schema（适合format schema等于table schema的场景，可自动派生出schema）来定义。

目前支持的JSON schema类型和Flink SQL类型如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70 12]

Missing Field Handling: By default, a missing JSON field is set to null. You can enable strict JSON parsing that will cancel the source (and query) if a field is missing.

Make sure to add the JSON format as a dependency.

需要在项目中添加JSON依赖：

<dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-json</artifactId>
        <version>${flink.version}</version>
    </dependency>

实例：

.withFormat(
      new Json()
      	// 可选。当某个field缺失的时候，是否导致失败，默认false。
        .failOnMissingField(true) 
    
        // 可选方式1。使用Flink数据类型定义，然后mapping映射解析为对应的type的类型信息来定义schema
        // flink的ROW对应JSON的object结构,String对应VARCHAR等
        .schema(Type.ROW(...))
    
        // 可选方式2。使用JSON schema来定义，可支持非常复杂和嵌套的数据结构
        .jsonSchema(
          "{" +
          " type: 'object'," +
          " properties: {" +
          " lon: {" +
          " type: 'number'" +
          " }," +
          " rideTime: {" +
          " type: 'string'," +
          		// 指定时间格式
          " format: 'date-time'" +
          " }" +
          " }" +
          "}"
        )
    
        // 可选方式3。可直接使用表的schema来解析
        // 适用于Flink Table Schema和JSON Schema一致时
        // 此时只需要定义Table Schema,就能确定字段名称、类型、位置顺序等
        .deriveSchema()
    )

## 4.8 Update Modes ##

### 4.8.1 概述 ###

流式查询中，需要声明怎么执行动态表和外部Connector之间的数据交换，有以下模式：

*  Append Mode（增）  
    仅交互`INSERT`操作数据
 *  Retract Mode （增删改）  
    交互`INSERT`（编码为`ADD`）和`DELETE`（编码为`RETRACT`）、`UPDATE`（对于修改前的行来说编码为`RETRACT`，对于修改后的新行来说编码为`ADD`）操作数据。
    
    与Upsert Mode相反，Retract Mode不能定义key。
    
    每个UPDATE操作由两条消息（RETRACT和ADD）组成，效率较低。
 *  Upsert Mode（增删改）  
    交互`UPSERT`（可编码`INSERT`和`UPDATE`）和`DELETE`操作数据。
    
    该模式需要一个唯一的key（可能是组合的），用来传播update事件。具体来说，外部连接器需要了解该唯一key属性，才能正确处理消息。
    
    Upsert Mode和Retract Mode都支持增删改，但不同是，Upsert Mode中的UPDATE事件使用单条UPSERT消息进行编码，而Retract Mode中的UPDATE由两条消息（RETRACT和ADD）组成，因此本模式效率更高。

### 4.8.2 例子 ###

.connect(...)
      .inAppendMode()    // otherwise: inUpsertMode() or inRetractMode()

每个connector支持哪些update mode，请参阅具体connector文档。

## 4.9 更多例子 ##

*  [Flink官方-table\_api][Flink_-table_api]  
    展示了如何实现一个对batch source的简单的Table API 查询，以及如何拓展为streaming source的持续查询。
 *  [Flink学习5-使用rowtime且分窗，Connector读取Kafka写入MySQL例子][Flink_5-_rowtime_Connector_Kafka_MySQL]
 *  [Apache Flink 各类关键数据格式读取/SQL支持][Apache Flink _SQL]
 *  [如何在 Flink 1.9 中使用 Hive？][Flink 1.9 _ Hive]
 *  [Flink SQL 解析复杂（嵌套）JSON][Flink SQL _JSON]

# 5 Connector #

## 5.1 [File System Connector][] ##

请点击[File System Connector][File System Connector 1]

## 5.2 [Kafka Connector][] ##

### 5.2.1 概述 ###

![在这里插入图片描述][20200729154957942.png]

Kafka Connector使得Flink可从Kafka中消费、写入数据。

*  依赖  
    Kafka 0.11以上版本后可采用统一的[flink-sql-connector-kafka\_2.11-1.11.0.jar][flink-sql-connector-kafka_2.11-1.11.0.jar]
 *  关于Flink分区和Kafka分区关系  
    可通过connector参数`sink.partitioner`指定：
    
     *  fixed  
        默认情况下，KafkaSink最多可以写入与其自身并行性（parallelism）一样多的Kafka分区，即每个并行的KafkaSink实例都固定写入一个Kafka分区。 如果实例数大于Kafka分区数，则空闲。
    
    即Flink生产者发送消息到Kafka时默认用的FlinkFixedPartitioner，如下图  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70 13]  
    这种方法有个明显的问题，即partition数大于task数时，多出来的partition不会有数据写入。partition扩容时，也需要重启Flinik程序。
    
     *  round-robin  
        循环分区器对于避免不平衡分区很有用， 但是，这将导致所有Flink实例与所有Kafka Broker节点之间的大量网络连接。
     *  自定义  
        为了将写操作分配到更多分区或自定义每行数据到分区的路由，可以提供自定义接收器分区程序，继承`FlinkKafkaPartitioner`。
 *  一致性保证  
    默认情况下，如果在启用检查点的时执行Flink，则KafkaSink会将具有至少一次(at least once)保证的数据提取到Kafka中。
    
    当使用Checkpoint时，提供exactly-once精准一次语义，具体原理是两阶段提交。
    
    除了使用Checkpoint，还可以通过设置参数`sink.semantic`来调节语义：
    
     *  at-least-once  
        默认值。保证至少一次，可能重复
     *  none  
        没有任何保障，可能多或少
     *  exactly-once  
        使用Kafka生产者事务来保证。注意此时必须同时设定Kafka Consumer事务隔离等级`isolation.level`为`read_committed`才会 读不到未提交的事务数据
 *  Kafka 0.10+的Timestamp属性  
    Kafka0.10开始，数据就带了一个`timestamp`作为元数据的一部分，该字段含义是数据写入Kafka的时间。该字段可用作Flink rowtime，请参考Java/Scala的`timestampsFromSource`方法。
 *  Kafka 0.11+版本  
    因为Flink1.7开始，Kafka Connector的定义就应该是独立于硬编码的Kafka version了，所以使用`.version("universal")`作为Kafka0.11开始的所有版本Kafka的通配符。
 *  Commit offset 策略
    
     *  不使用Checkpoint时，由`enable.auto.commit`, `auto.commit.interval.ms`一起决定 commit 行为，每隔一段时间向 kafka commit 一次 offset。
     *  使用Checkpoint时，kafka consumer 将 offset commit 到 checkpoint state 中。
 *  Consumer 分配策略  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70 14]  
    如上图，不同 topic 之间的 startIndex 是随机的，故解决了多 topic 负载均衡问题。
    
    **注意，Consumer 无法感应获取 Kafka topic 新增的Partition！也就是说，某topic扩容Partition后必须重启Flink程序。**
 *  Changelog Source  
    Flink支持将Kafka作为`Changelog Source`。
    
    只要使用CDC工具从其他数据库捕获变动event放入Kafka，则可以使用`cdc format`来解释kafka中数据，转为`INSERT/UPDATE/DELETE`来放入Flink SQL系统中。目前支持[debezium-json][] 和 [canal-json][]。
 *  数据类型映射  
    Kafka按[format][]将数据序列化为字节，没有数据类型和schema。反序列化也是根据`format`.
 *  元数据  
    可直接通过SQL获取Kafka元数据，应用到定义的列中  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70 15]  
    以上R表示可读，W表示可写。只读的字段必须在定义时加上`VIRTUAL`，以在`INSERT INTO`时排除这些字段。

### 5.2.2 CDC - Changelog Source ###

如果Kafka数据来源于使用CDC工具从其他数据库捕获的，则你可以使用`CDC format`来将数据解析为Flink SQL中的`INSERT/UPDATE/DELETE`，然后继续处理，如同步数据到其他目的地等。

目前支持CDC格式如下：

*  [debezium-json][]
 *  [canal-json][]

相关文档：

*  [阿里伍翀讲解CDC][CDC]
 *  [Flink SQL CDC 上线！我们总结了 13 条生产实践经验][Flink SQL CDC _ 13]

### 5.2.3 配置 ###

<table> 
 <thead> 
  <tr> 
   <th>配置项</th> 
   <th>Required</th> 
   <th>Default</th> 
   <th>Type</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>connector</td> 
   <td>required</td> 
   <td>(none)</td> 
   <td>String</td> 
   <td>‘kafka’, ‘kafka-0.11’, ‘kafka-0.10’.</td> 
  </tr> 
  <tr> 
   <td>topic</td> 
   <td>required</td> 
   <td>none)</td> 
   <td>String</td> 
   <td>Topic name from which the table is read.</td> 
  </tr> 
  <tr> 
   <td>properties.bootstrap.servers</td> 
   <td>required</td> 
   <td>(none)</td> 
   <td>String</td> 
   <td>Kafka brokers，逗号分隔.</td> 
  </tr> 
  <tr> 
   <td>properties.group.id</td> 
   <td>required by source</td> 
   <td>(none)</td> 
   <td>String</td> 
   <td>Kafka source consumer id<br> optional for Kafka sink.</td> 
  </tr> 
  <tr> 
   <td>format</td> 
   <td>required</td> 
   <td>(none)</td> 
   <td>String</td> 
   <td>deserialize/serialize Kafka messages的<a href="https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/dev/table/connectors/formats/" rel="nofollow">format</a>. 可填 ‘csv’, ‘json’, ‘avro’, ‘debezium-json’ , ‘canal-json’.</td> 
  </tr> 
  <tr> 
   <td>scan.startup.mode</td> 
   <td>optional</td> 
   <td>group-offsets</td> 
   <td>String</td> 
   <td>Kafka consumer Startup mode, valid values： <br>‘earliest-offset’<br> ‘latest-offset’<br> ‘group-offsets’(从zk/kafka里保存的指定consumer group的offset开始消费)<br> ‘timestamp’(从用户定义的<code>scan.startup.timestamp-millis</code>中时间戳开始消费) <br> ‘specific-offsets’(从用户定义的<code>scan.startup.specific-offsets</code>开始消费)</td> 
  </tr> 
  <tr> 
   <td>scan.startup.specific-offsets</td> 
   <td>optional</td> 
   <td>(none)</td> 
   <td>String</td> 
   <td>‘specific-offsets’ startup mode时指定每个分区开始消费的offset, 如’partition:0,offset:42;partition:1,offset:300’.</td> 
  </tr> 
  <tr> 
   <td>scan.startup.timestamp-millis</td> 
   <td>optional</td> 
   <td>(none)</td> 
   <td>Long</td> 
   <td>‘timestamp’ startup mode时指定消费起始毫秒级时间戳</td> 
  </tr> 
  <tr> 
   <td>sink.partitioner</td> 
   <td>optional</td> 
   <td>(none)</td> 
   <td>String</td> 
   <td>Flink’s partitions 映射到输出的 Kafka’s partitions方式. 合法值如下：<br>fixed: 每个 Flink partition 最多输入一个Kafka partition<br>round-robin: 每个Flink partition轮询方式映射到所有Kafka partitions<br>自定义 FlinkKafkaPartitioner subclass: 如 ‘org.mycompany.MyPartitioner’.</td> 
  </tr> 
 </tbody> 
</table>

### 5.2.4 例子 ###

*  1.11.0之前的例子，比较复杂：
    
        .connect(
          new Kafka()
          	// 必填。版本号（"0.8", "0.9", "0.10", "0.11", "universal"）
            .version("0.11")   
            // 必填。订阅的topic
            .topic("student_info")  
        
            // Kafka连接属性
            .property("zookeeper.connect", "localhost:2181")
            .property("bootstrap.servers", "localhost:9092")
            .property("group.id", "testGroup")
        
            // 可选如下。指定无法找到group对应的offset时，从哪儿开始消费。
            .startFromEarliest()
            .startFromLatest()
            .startFromSpecificOffsets(...)
        
            // 当Flink分区往kafka分区写入数据时需要配置。 
            // 默认。每个Flink分区最多被分配到一个Kafka分区
            .sinkPartitionerFixed()
            // 一个Flink分区会以轮询方式发送到Kafka各个分区 
            .sinkPartitionerRoundRobin()    
            // 自定义FlinkKafkaPartitioner
            .sinkPartitionerCustom(MyCustom.class)    
        )
 *  1.11.0的例子，简化了很多：
    
        CREATE TABLE kafka_source_table (
         user_id BIGINT,
         item_id BIGINT,
         category_id BIGINT,
         behavior STRING,
         message ROW<log_create_time BIGINT>,
         ts_field AS TO_TIMESTAMP(FROM_UNIXTIME(message.log_create_time)),
        ) WITH (
         'connector' = 'kafka',
         'topic' = 'user_behavior',
         'properties.bootstrap.servers' = 'localhost:9092',
         'properties.group.id' = 'testGroup',
         'format' = 'json',
         'scan.startup.mode' = 'group-offsets'
        )
 *  1.12.0的例子，支持获取Kafka元数据
    
        CREATE TABLE KafkaTable (
          `event_time` TIMESTAMP(3) METADATA FROM 'value.source.timestamp' VIRTUAL,  -- from Debezium format
          `origin_table` STRING METADATA FROM 'value.source.table' VIRTUAL, -- from Debezium format
          `partition_id` BIGINT METADATA FROM 'partition' VIRTUAL,  -- from Kafka connector
          `offset` BIGINT METADATA VIRTUAL,  -- from Kafka connector
          `user_id` BIGINT,
          `item_id` BIGINT,
          `behavior` STRING
        ) WITH (
          'connector' = 'kafka',
          'topic' = 'user_behavior',
          'properties.bootstrap.servers' = 'localhost:9092',
          'properties.group.id' = 'testGroup',
          'scan.startup.mode' = 'earliest-offset',
          'value.format' = 'debezium-json'
        );

## 5.3 [Upsert Kafka][] ##

Flink 1.12以前版本仅支持append only元素写入，而当我们产生聚合元素的时候无法使用kafka sink.

Flink 1.12特地新增`Upsert Kafka SQL Connector`解决此问题。

作为Source时：

*  生产`changelog`流
    
     *  有value的表示`UPDATE`
     *  value为null的表示`DELETE`
 *  作为Sink时，可消费`changelog`流
    
     *  `INSERT`/`UPDATE_AFTER`作为普通消息输出到Kafka
     *  `DELETE`会作为`null`值数据以示这是该key对应数据的墓碑
     *  由于Flink按照用户定义的主键列值进行消息分区，可保证该分区数据有序，同Key数据也会发到相同分区

官方示例：

CREATE TABLE pageviews_per_region (
      user_region STRING,
      pv BIGINT,
      uv BIGINT,
      PRIMARY KEY (region) NOT ENFORCED
    ) WITH (
      'connector' = 'upsert-kafka',
      'topic' = 'pageviews_per_region',
      'properties.bootstrap.servers' = '...',
      'key.format' = 'avro',
      'value.format' = 'avro'
    );
    
    CREATE TABLE pageviews (
      user_id BIGINT,
      page_id BIGINT,
      viewtime TIMESTAMP,
      user_region STRING,
      WATERMARK FOR viewtime AS viewtime - INTERVAL '2' SECOND
    ) WITH (
      'connector' = 'kafka',
      'topic' = 'pageviews',
      'properties.bootstrap.servers' = '...',
      'format' = 'json'
    );
    
    -- calculate the pv, uv and insert into the upsert-kafka sink
    INSERT INTO pageviews_per_region
    SELECT
      user_region,
      COUNT(*),
      COUNT(DISTINCT user_id)
    FROM pageviews
    GROUP BY user_region;

## 5.4 Elasticsearch Connector ##

## 5.5 [JDBC Connector][JdbcCatalog documentation] ##

### 5.5.1 概述 ###

![在这里插入图片描述][20200729162621147.png]  
可以通过 JDBC Connector读写JDBC数据源。

目前支持[Mysql][]、PostgreSQL、Derby等，需要下载对应Jar包放入`$FLINK_HOME/lib`下

需要添加的依赖为

*  [flink-connector-jdbc\_2.11-1.11.0.jar][flink-connector-jdbc_2.11-1.11.0.jar]
 *  特定的数据库所需的driver  
    比如mysql-connector-java-5.1.47.jar

更多配置及数据库和Flink的类型映射请参考[官网][JdbcCatalog documentation]。

### 5.5.2 Streaming Mode ###

在流场景可支持

*  Append（适用于只交互INSERT信息场景）  
    DDL不提供主键时
 *  Upsert（使用唯一key，适用于交互增删改信息场景）模式。  
    DDL提供主键时

Flink自动从流式查询中提取有效key。 例如，查询`SELECT a, b, c FROM t GROUP BY a, b`定义了字段a和b的组合键。

如果将JDBC表用作`upsert sink`，请确保查询的key是数据库的唯一键之一或是主键。 这样才可以保证输出结果符合预期。

### 5.5.3 Temporary Join与查询缓存 ###

JDBC Connector可以被视为`lookup source`来做临时join。

当前，只支持同步模式的\`lookup source。

可使用`loopup source cache`来提高临时join的性能（**默认不开启缓存**），具体原理是每个TaskManager先查询本节点持有的缓存而不是直接查询数据库本身，当目标不在缓存时再请求远程数据库，并将返回的结果更新到缓存。

**但务必注意，使用缓存时数据可能不是最新，这需要使用者进行权衡**！

loopup source cache相关选项有：

当数据作为维表时，如果使用缓存，需要注意：

*  指定缓存后Join维表时拿到的可能是过期数据，其实已经被改变。但设置缓存过期时间太短会造成频繁查库增加性能开销。而将缓存条数设置过大可能打爆内存（可以适当调大内存，但不是根本方式）。所以，缓存的大小和时间配置需要权衡和实验。不过通常来说，维表中的数据应该缓慢变化的。
 *  Flink Sql 1.9 原生支持的维表关联只支持同步模式，如果需要异步模式或者想用其他的第三方存储只能够自己去实现。

### 5.5.4 数据写入缓存与Flush ###

Flink JDBC Connector会将数据先放入内存，到达触发条件时再flush到jdbc数据库中。相关配置如下：

*  sink.buffer-flush.interval  
    默认1秒，可选的。当flush间隔时间超过此值时，会有一个异步线程将数据刷入数据库表。
    
    可设为0禁用。
    
    可将`sink.buffer-flush.max-rows`设为0 ，本值设为大于0的数，以实现完全异步的缓冲处理。
 *  sink.buffer-flush.max-rows  
    默认100行，可选的。当数据条数（包括增、删、改）累积到该值时，才会将数据输入数据库表.
    
    可设为0禁用。
 *  sink.max-retries  
    默认3，可选的。当某批数据写入数据表失败时的最大重试次数。

### 5.5.5 主键 ###

1.11可使用DDL来定义主键了，定义后采用`UPSERT`流模式，否则为`APPEND`模式。

*  采用`UPSERT`流模式时，是根据主键来插入或更新记录，由此来保证幂等性。
    
    推荐将设置主键，且需要确保主键是该数据表的唯一键或主键。
 *  `APPEND`模式下，所有数据都作为`INSERT`消息处理，当底层数据库表的主键或唯一键冲突时会导致插入失败。

### 5.5.6 幂等写入 ###

JDBCSink在DDL中定义了主键列的情况，会使用UPSERT语义而不是INSERT语义。

UPSERT语义可使得当底层数据库存在唯一键约束时，原子性得插入一行或更新已有行，提供了幂等性。

应用场景：

*  这样一来，即使发生失败需要从最后成功的Checkpoint重启导致重复处理数据，也能保证幂等，不会出现数据重复。
 *  数据源头生成相同主键数据，需要更新

没有标准upsert语法，根据底层数据库不同而不同，比如MySQL语法如下：

INSERT .. ON DUPLICATE KEY UPDATE ..

### 5.5.7 Partitioned Scan ###

目的是加速从JDBC数据表读取数据，手段是使用并行Source task多实例。

以下任一`scan partition`选项被指定，则所有`scan partition`选项都必须被指定。作用是在并行读取数据时指定将表分区的方式。以下这些选项是当从表读数据时使用：

*  scan.partition.column  
    线程读取表时分区的列，只能是数值、日期或timestamp
 *  scan.partition.num  
    partitions数，决定了多线程并行读取表时指定怎么对表分区.
 *  scan.partition.lower-bound  
    用于确定分区步幅下界（不是用于过滤表中的行）
 *  scan.partition.upper-bound  
    用于确定分区步幅上界

### 5.5.8 例子 ###

*  1.11以前
    
        CREATE TABLE MyUserTable (
          ...
        ) WITH (
          'connector.type' = 'jdbc', -- 必填: 指定动态表类型为jdbc
          
          'connector.url' = 'jdbc:mysql://localhost:3306/flink-test', -- 必填: JDBC DB url
          
          'connector.table' = 'jdbc_table_name',  -- 必填: jdbc table name
        
          -- 可选的: JDBC driver ，不填时从url中派生
          'connector.driver' = 'com.mysql.jdbc.Driver',
        
          -- 可选的: jdbc user name and password
          'connector.username' = 'name',
          'connector.password' = 'password',
          
          -- **以下这些选项是当从表读数据时使用，是一些可选的浏览选项**
          -- 注意，只要以下任何一项值被填写，那就需要填写所有选项值
          'connector.read.partition.column' = 'column_name', -- optional: 多线程读取表时分区的列，只能是数值、日期或timestamp.
          'connector.read.partition.num' = '50', -- optional: partitions数，多线程并行读取表时指定怎么对表分区.
          'connector.read.partition.lower-bound' = '500', -- optional: 用于确定分区步幅下界（不是用于过滤表中的行）
          'connector.read.partition.upper-bound' = '1000', -- optional: 用于确定分区步幅上界
          -- optional, 默认0，给读取者一个关于每轮读取数据行数的提示，如果设为0则提示被忽略
          'connector.read.fetch-size' = '100',
        
          -- **以下为temporary join时的lookup选项**
          'connector.lookup.cache.max-rows' = '5000', -- optional
          'connector.lookup.cache.ttl' = '10s', -- optional
          'connector.lookup.max-retries' = '3', -- optional
        
          -- **以下为写入jdbc表时使用的sink选项**
          'connector.write.flush.max-rows' = '5000', -- optional
          'connector.write.flush.interval' = '2s', -- optional
          'connector.write.max-retries' = '3' -- optional
        )
 *  1.11
    
        -- register a MySQL table 'users' in Flink SQL
        CREATE TABLE MyUserTable (
          id BIGINT,
          name STRING,
          age INT,
          status BOOLEAN,
          PRIMARY KEY (id) NOT ENFORCED
        ) WITH (
           'connector' = 'jdbc',
           'url' = 'jdbc:mysql://localhost:3306/mydatabase',
           'table-name' = 'users'
        );
        
        -- write data into the JDBC table from the other table "T"
        INSERT INTO MyUserTable
        SELECT id, name, age, status FROM T;
        
        -- scan data from the JDBC table
        SELECT id, name, age, status FROM MyUserTable;
        
        -- temporal join the JDBC table as a dimension table
        SELECT * FROM myTopic
        LEFT JOIN MyUserTable FOR SYSTEM_TIME AS OF myTopic.proctime
        ON myTopic.key = MyUserTable.id;

## 5.6 [Hive][Hive 2] ##

请点击[HiveConnector][]

## 5.7 自定义Source Sink ##

### 5.7.1 概述 ###

请参考[User-defined Sources & Sinks][Connector]

### 5.7.2 自定义Source ###

#### 5.7.2.1 ScanTableSource ####

##### 5.7.2.1.1 概述 #####

ChangeLog，Source在batch场景发送有界的、insert-only流；Streaming场景发送无界的、insert-only流；CDC(Change Data Capture)场景发送无界/有界的、可包含增删改的记录的流。

从外部数据中扫描整张表的所有行。可包含增删改的行，所以可被用来读取changelog。

`getChangelogMode`返回`ChangelogMode`，表示运行时可能返回的Changelog类型集合。

如果Source有一些额外的能力，可以实现`org.apache.flink.table.connector.source.abilities`包内的接口，如`SupportsProjectionPushDown`投影下推。

ScanTableSource生产的记录必须是`org.apache.flink.table.data.RowData`，用来在Flink内部传递

其中定义了一个重要的方法`getScanRuntimeProvider`，需要得到一个`ScanRuntimeProvider`，可通过`InputFormatProvider.of(InputFormat<RowData, ?> inputFormat)`得到，如`HBaseDynamicTableSource`：

@Override
    public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderContext) { 
    	return InputFormatProvider.of(new HBaseRowDataInputFormat(conf, tableName, hbaseSchema, nullStringLiteral));
    }

也可通过`SourceFunctionProvider of(SourceFunction<RowData> sourceFunction, boolean isBounded)`得到，如`KafkaDynamicSourceBase`：

@Override
    public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderContext) { 
    	DeserializationSchema<RowData> deserializationSchema =
    			this.decodingFormat.createRuntimeDecoder(runtimeProviderContext, this.outputDataType);
    	// Version-specific Kafka consumer
    	FlinkKafkaConsumerBase<RowData> kafkaConsumer =
    			getKafkaConsumer(topic, properties, deserializationSchema);
    	return SourceFunctionProvider.of(kafkaConsumer, false);
    }

##### 5.7.2.1.2 HBaseDynamicTableSource #####

public class HBaseRowDataInputFormat extends AbstractTableInputFormat<RowData> { 
    	public HBaseRowDataInputFormat(
    			org.apache.hadoop.conf.Configuration conf,
    			String tableName,
    			HBaseTableSchema schema,
    			String nullStringLiteral) { 
    		super(conf);
    		this.tableName = tableName;
    		this.schema = schema;
    		this.nullStringLiteral = nullStringLiteral;
    	}
    }

abstract class AbstractTableInputFormat<T> extends RichInputFormat<T, TableInputSplit> { 
    	public AbstractTableInputFormat(org.apache.hadoop.conf.Configuration hConf) { 
    		serializedConfig = HBaseConfigurationUtil.serializeConfiguration(hConf);
    	}
    }

1.  HBaseRowDataInputFormat\#configure  
    创建了serde、Connection、HTable、Scan(此时无startRow和stopRow)
2.  createInputSplits  
    获取该HTable的每个Region Start和End Key，为每个Region创建一个`TableInputSplit`（包含spilitId, hosts, tableName(), splitStart, splitStop），构成`TableInputSplit[]`
3.  每个并行的input task创建一个实例，并调用`open(T split)`方法为该并行实例负责的spilit初始化。
    
        // set scan range
        currentRow = split.getStartRow();
        scan.setStartRow(currentRow);
        scan.setStopRow(split.getEndRow());
        // 准备scanner进行scan
        resultScanner = table.getScanner(scan);
        // reachedEnd方法会通过该变量判断是否input读取完成
        endReached = false;
        scannedRows = 0;
4.  读取数据
    
        // 这里T就是RowData
        public T nextRecord(T reuse) throws IOException { 
        	Result res;
        	try { 
        		// scan出的一条数据
        		res = resultScanner.next();
        	} catch (Exception e) { 
        		resultScanner.close();
        		//workaround for timeout on scan
        		LOG.warn("Error after scan of " + scannedRows + " rows. Retry with a new scanner...", e);
        		// 从当前已读位置的下一个位置继续读
        		scan.withStartRow(currentRow, false);
        		resultScanner = table.getScanner(scan);
        		res = resultScanner.next();
        	}
        	if (res != null) { 
        		scannedRows++;
        		// 记录已读位置
        		currentRow = res.getRow();
        		// 组装为RowData返回
        		return mapResultToOutType(res);
        	}
        	// 没读到数据时就说明已经读取完input
        	endReached = true;
        	return null;	
        }
5.  close  
    当`reachedEnd`后说明 一个input split读取完成后调用本方法，被用来关闭channel和流，释放资源。
    
        currentRow = null;
        try { 
        	if (resultScanner != null) { 
        		resultScanner.close();
        	}
        } finally { 
        	resultScanner = null;
        }
6.  可能会再次open来读取下一个split

### 5.7.3 自定义Sink ###

### 5.7.4 自定义TableFactory ###

## 5.8 [DataGen][] ##

### 5.8.1 概述 ###

DataGen是一个调试用的Connector，可以配合使用[Computed Column syntax][]来生成简单类型数据。

### 5.8.2 数据生成器 ###

目前有两类：

*  RandomGenerator  
    默认选项，是一个无界数据生成器。
    
    可指定随机值的最大和最小区间。
    
    对于字符串类型数据（char/varchar/string），可指定长度。
 *  SequenceGenerator  
    是一个有界数据生成器。
    
    可指定序列的开始和结束值，当Sequence抵达结束值时，读取结束。

### 5.8.3 配置 ###

转自官网：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70 16]

### 5.8.4 例子 ###

CREATE TABLE datagen (
     f_sequence INT,
     f_random INT,
     f_random_str STRING,
     -- 以当前本地时区获取时间戳
     ts AS localtimestamp,
     WATERMARK FOR ts AS ts
    ) WITH (
     'connector' = 'datagen',
    
     -- optional options --
     -- task每秒生成5条数据
     'rows-per-second'='5',
     -- f_sequence字段使用SequenceGenerator，从1到100顺序生成
     'fields.f_sequence.kind'='sequence',
     'fields.f_sequence.start'='1',
     'fields.f_sequence.end'='1000',
     -- f_random字段使用RandomGenerator，从1到1000随机生成
     'fields.f_random.min'='1',
     'fields.f_random.max'='1000',
     -- f_random_str使用RandomGenerator，指定字符串长度为10
     'fields.f_random_str.length'='10'
    )

## 5.9 HBase Connector ##

# 6 Function #

参考[Functions][]

## 6.1 内置函数 ##

## 6.2 UDF ##

# 7 通用配置 #

参考:[Configuration][Link 3]

## 7.1 checkpoint ##

pipeline.time-characteristic EventTime
    execution.checkpointing.interval 120000
    execution.checkpointing.min-pause 60000
    execution.checkpointing.timeout 60000
    execution.checkpointing.externalized-checkpoint-retention RETAIN_ON_CANCELLATION
    execution.savepoint.path hdfs:/tmp/flink-checkpoints/xxx/chk-yyy

# 8 调优 #

## 8.1 概述 ##

目前官网只有流式聚合调优一项。

## 8.2 流式聚合调优调优 ##

参考[Streaming Aggregation][]

### 8.2.1 概述 ###

Flink SQL允许用户直接用SQL来定义效率高的流式分析程序，已经做了大量工作来做查询优化和算子调优，但有一些选项需要手动开启进行调优。

注意：

*  当前这里的调优只支持Blink Planner。
 *  当前流式聚合调优只支持[无界聚合][Link 8]，未来会支持对窗口聚合的调优。

默认状况下，无界聚合算子逐条处理输入记录：

1.  从状态读取累加器`accumulator`
2.  从累加器中累加/撤回记录
3.  将累加器写回位于StateBackend的状态中
4.  下一条记录又从步骤1到3进行处理

通过了解无界聚合处理步骤，可以看到这样的处理模式会增加StateBackend（尤其是硬盘上的RocksDBStatebackend）的开销。此外，数据倾斜在生产中非常常见，更会加重此问题，使得job容易发生反压。

### 8.2.2 MiniBatch聚合 ###

#### 8.2.2.1 概述 ####

MiniBatch聚合核心思想是在聚合算子内部缓冲区缓存一组输入的数据，当触发计算时，同一个key的所有记录只需要一个操作即可访问状态了。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 2]

#### 8.2.2.2 特点 ####

*  这和传统无界聚合方法有很大不同，这样做能很大程度减少状态开销，提升吞吐。
 *  但同时也会因为需要缓存输入到聚合算自内部而带来额外开销和延时，而不是立刻处理每条输入记录，其实就是吞吐和延时之间的一个折中

#### 8.2.2.3 配置 ####

MiniBatch聚合默认关闭。

要开启，需要配置：

*  table.exec.mini-batch.enabled  
    是否允许MiniBatch聚合。一旦开启，必须配置一下两个选项。
 *  table.exec.mini-batch.allow-latency  
    最大允许的MiniBatch聚合延迟时间
 *  table.exec.mini-batch.size  
    最大允许的MiniBatch聚合缓存数据大小，级别是每个聚合算子task

配置方法：

// instantiate table environment
    val tEnv: TableEnvironment = ...
    
    // access flink configuration
    val configuration = tEnv.getConfig().getConfiguration()
    // set low-level key-value options
    configuration.setString("table.exec.mini-batch.enabled", "true") // enable mini-batch optimization
    configuration.setString("table.exec.mini-batch.allow-latency", "5 s") // use 5 seconds to buffer input records
    configuration.setString("table.exec.mini-batch.size", "5000") // the maximum number of records can be buffered by each aggregate operator task

### 8.2.3 LocalGlobalBatch聚合 ###

#### 8.2.3.1 概述 ####

LocalGlobalBatch聚合主要目的是解决数据倾斜，方法是将`group by`聚合拆解为两个阶段：

1.  先在上游做本地聚合
2.  在下游做全局聚合

这个思路类似于MR中的MapCombine + Reduce Combine两个阶段。

比如以下SQL

SELECT color, sum(id)
    FROM T
    GROUP BY color

有可能上游`color`字段数据已经倾斜，因此下游的某几个聚合算子实例必须处理比其他实例更多的数据，这就造成了数据热点，即数据倾斜。

此时就可使用 LocalGlobalBatch聚合，在上游算子将同key数据进行本地聚合，放入同一个累加器，只将聚合后的累加器发送到下游聚合算子，而不是原始的所有记录，这样大大减少网络shuffle时的开销以及状态访问开销。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 3]

每次本地聚合时累加的输入记录数目取决于`mini-batch interval`，也就是说LocalGlobalBatch聚合前提是`mini-batch`优化必须已经开启。

#### 8.2.3.2 特点 ####

大大减少网络shuffle时的开销以及状态访问开销

#### 8.2.3.3 配置 ####

LocalGlobalBatch聚合默认关闭。

要开启，需要配置：

*  table.exec.mini-batch.enabled  
    是否允许MiniBatch聚合。一旦开启，必须配置一下两个选项。
 *  table.optimizer.agg-phase-strategy  
    TWO\_PHASE
    
    默认AUTO（根据cost来决定采用两阶段还是一阶聚段合），还可填TWO\_PHASE(两阶段聚合)、ONE\_PHASE(一阶段聚合)

配置方法：

// instantiate table environment
    val tEnv: TableEnvironment = ...
    
    // access flink configuration
    val configuration = tEnv.getConfig().getConfiguration()
    // set low-level key-value options
    configuration.setString("table.exec.mini-batch.enabled", "true") // local-global aggregation depends on mini-batch is enabled
    configuration.setString("table.exec.mini-batch.allow-latency", "5 s")
    configuration.setString("table.exec.mini-batch.size", "5000")
    configuration.setString("table.optimizer.agg-phase-strategy", "TWO_PHASE") // enable two-phase, i.e. local-global aggregation

### 8.2.4 拆分DistinctBatch聚合 ###

#### 8.2.4.1 概述 ####

LocalGlobalBatch聚合优化可在常规聚合场景有效消除数据倾斜，如SUM, COUNT, MAX, MIN, AVG，但不适合处理`DISTINCT`聚合！

比如计算每天的用户UV:

SELECT day, COUNT(DISTINCT user_id)
    FROM T
    GROUP BY day

当distinct key（比如这里的user\_id，因为distinct的是`user_id`，而不是`day`）稀疏时，`COUNT DISTINCT`一般不能有效聚合而减少记录数，即使开启了LocalGlobalBatch聚合提升也不大。因为key稀疏场景下在`COUNT DISTINCT`后累加器依然持有大部分原始记录（比如这里在本地聚合后，只是得到了去重后的`user_id`而已），所以全局聚合阶段依然存在瓶颈。比如以上例子中，大多数负载高的累加器由一个task处理，也就是属于同一天的被一个task处理。

针对`COUNT DISTINCT`以上问题，就有了拆分DistinctBatch聚合优化思想，将`COUNT DISTINCT`拆分为两个层级：

1.  首次聚合先通过`group key`和额外自定义的`bucket key`进行shuffle，并进行聚合计算。  
    `bucket key`是通过`HASH_CODE(distinct_key) % BUCKET_NUM`计算，`BUCKET_NUM`默认1024，可通过`table.optimizer.distinct-agg.split.bucket-num`调整。
2.  第二次聚合通过原始`group key`进行shuffle，并使用`SUM`聚合来自不同 bucket 的 `COUNT DISTINCT` 值

这个过程如下图，这里不同颜色表示热点`group key`，类似我们例子的`day`；字母表示user\_id：左图为本地全局聚合，右图为拆分DistinctBatch聚合：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JhaWNob3VmZWk5MA_size_16_color_FFFFFF_t_70_pic_center 4]  
可以看到：

*  开启本地全局聚合计算`COUNT DISTINCT`时，本地聚合阶段其实并没有减少太多数据量，导致传输到右上聚合算子时的数据过多，造成数据热点和倾斜，成为瓶颈。
 *  开启拆分DistinctBatch聚合计算`COUNT DISTINCT`时，首先根据group key即day和distinct key即user\_id计算hash并对bucket取余后得到BucketKey联合进行shuffle，这个时候就已经将热点数据打散到各个第一阶段聚合算子上了。这一阶段就会计算`COUNT DISTINCT`，输出就是唯一值的数量；
    
    随后根据group key即day进行第二阶段聚合，此时由于第一阶段已经将相同group by数据聚合计算，所以第二阶段做的工作就少了很多了。此时只需要将第一阶段的唯一值SUM求和即可。

相同distinct key的行会在同一个Bucket内计算，所以能这样分阶段聚合。第一阶段聚合的意义是分担热点的`group key`，比如以上例子时可将最新一天的数据shuffle到不同task计算，有效解决 `COUNT DISTINCT` 时的数据倾斜和数据热点问题。

在拆分DistinctBatch聚合优化后，以上查询会被自动重写：

SELECT day, SUM(cnt)
    FROM (
        SELECT day, COUNT(DISTINCT user_id) as cnt
        FROM T
        GROUP BY day, MOD(HASH_CODE(user_id), 1024)
    )
    GROUP BY day

以上为做简单的例子，但其实Flink支持拆分更复杂的聚合查询，比如，多个具有不同 distinct key （例如 COUNT(DISTINCT a), SUM(DISTINCT b) ）的 distinct 聚合，可以与其他非 distinct 聚合（例如 SUM、MAX、MIN、COUNT ）配合使用。

需要注意的是，目前聚合拆分优化不支持包含用户定义的 AggregateFunction 的聚合运算。

#### 8.2.4.2 特点 ####

适用于distinct聚合时出现数据倾斜的场景。

#### 8.2.4.3 配置 ####

配置方法：

// instantiate table environment
    val tEnv: TableEnvironment = ...
    
    tEnv.getConfig         // access high-level configuration
      .getConfiguration    // set low-level key-value options
      .setString("table.optimizer.distinct-agg.split.enabled", "true")  // enable distinct agg split