Flink 1.11新版本解读：更易用的流批一体SQL引擎

陈玉兆（玉兆） 2020-07-27 11:12:43

许多的数据科学家、分析师和 BI 用户依赖交互式 SQL 查询分析数据。Flink SQL 是 Flink 的核心模块之一。作为一个分布式的 SQL 查询引擎。Flink SQL 提供了各种异构数据源的联合查询。开发者可以很方便地在一个程序中通过 SQL 编写复杂的分析查询。通过 CBO 优化器、列式存储、和代码生成技术，Flink SQL 拥有非常高的查询效率。同时借助于 Flink runtime 良好的容错和扩展性，Flink SQL 可以轻松处理海量数据。

在保证优秀性能的同时，易用性是 1.11 版本 Flink SQL 的重头戏。易用性的提升主要体现在以下几个方面：

更方便的追加或修改表定义；
灵活的声明动态的查询参数；
加强和统一了原有 TableEnv 上的 SQL 接口；
简化了 connector 的属性定义；
对 Hive 的 DDL 做了原生支持；
加强了对 python UDF 的支持。

下面逐一为大家介绍 ~

Create Table Like

在生产中，用户常常有调整现有表定义的需求。例如用户想在一些外部的表定义（例如 Hive metastore）基础上追加 Flink 特有的一些定义比如 watermark。在 ETL 场景中，将多张表的数据合并到一张表，目标表的 schema 定义其实是上游表的合集，需要一种方便合并表定义的方式。

从 1.11 版本开始，Flink 提供了 LIKE 语法，用户可以很方便的在已有的表定义上追加新的定义。

例如我们可以使用下面的语法给已有表 base_table 追加 watermark 定义：

CREATE [TEMPORARY] TABLE base_table (

id BIGINT,

name STRING,

tstmp TIMESTAMP,

PRIMARY KEY(id)

) WITH (

'connector': 'kafka'

)

CREATE [TEMPORARY] TABLE derived_table (

WATERMARK FOR tstmp AS tsmp - INTERVAL '5' SECOND

)

LIKE base_table;

这里 derived_table 表定义等价于如下定义：

CREATE [TEMPORARY] TABLE derived_table (

id BIGINT,

name STRING,

tstmp TIMESTAMP,

PRIMARY KEY(id),

WATERMARK FOR tstmp AS tsmp - INTERVAL '5' SECOND

) WITH (

‘connector’: ‘kafka’

)

对比之下，新的语法省去了重复的 schema 定义，用户只需要定义追加属性，非常方便简洁。

多属性策略

有的小伙伴会问，原表和新表的属性只是新增或追加吗？如果我想覆盖或者排除某些属性该如何操作？这是一个好问题，Flink LIKE 语法提供了非常灵活的表属性操作策略。

LIKE 语法支持使用不同的 keyword 对表属性分类：

ALL：完整的表定义；
CONSTRAINTS: primary keys, unique key 等约束；
GENERATED: 主要指计算列和 watermark；
OPTIONS: WITH (...) 语句内定义的 table options；
PARTITIONS: 表分区信息。

在不同的属性分类上可以追加不同的属性行为：

INCLUDING：包含（默认行为）；
EXCLUDING：排除；
OVERWRITING：覆盖。

下面这张表格说明了不同的分类属性允许的行为：

例如下面的语句:

CREATE [TEMPORARY] TABLE base_table (

id BIGINT,

name STRING,

tstmp TIMESTAMP,

PRIMARY KEY(id)

) WITH (

'connector': 'kafka',

'scan.startup.specific-offsets': 'partition:0,offset:42;partition:1,offset:300',

'format': 'json'

)

CREATE [TEMPORARY] TABLE derived_table (

WATERMARK FOR tstmp AS tsmp - INTERVAL '5' SECOND

)

WITH (

'connector.starting-offset': '0'

)

LIKE base_table (OVERWRITING OPTIONS, EXCLUDING CONSTRAINTS);

等价的表属性定义为：

CREATE [TEMPORARY] TABLE derived_table (

id BIGINT,

name STRING,

tstmp TIMESTAMP,

WATERMARK FOR tstmp AS tsmp - INTERVAL '5' SECOND

) WITH (

'connector': 'kafka',

'scan.startup.specific-offsets': 'partition:0,offset:42;partition:1,offset:300',

'format': 'json'

)

细节参见：https://ci.apache.org/projects/flink/flink-docs-master/dev/table/sql/create.html#create-table

Dynamic Table Options

在生产中，调整参数是一个常见需求，很多的时候是临时修改（比如通过终端查询和展示），比如下面这张 Kafka 表：

create table kafka_table (

id bigint,

age int,

name STRING

) WITH (

'connector' = 'kafka',

'topic' = 'employees',

'scan.startup.mode' = 'timestamp',

'scan.startup.timestamp-millis' = '123456',

'format' = 'csv',

'csv.ignore-parse-errors' = 'false'

)

在之前的版本，如果用户有如下需求：

用户需要指定特性的消费时间戳，即修改 scan.startup.timestamp-millis 属性；
用户想忽略掉解析错误，需要将 format.ignore-parse-errors 改为 true。

只能使用 ALTER TABLE 这样的语句修改表的定义，从 1.11 开始，用户可以通过动态参数的形式灵活地设置表的属性参数，覆盖或者追加原表的 WITH (...) 语句内定义的 table options。

基本语法为：

table_name /*+ OPTIONS('k1'='v1', 'aa.bb.cc'='v2') */

OPTIONS 内的键值对会覆盖原表的 table options，用户可以在各种 SQL 语境中使用这样的语法，例如：

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;

动态参数的使用没有语境限制，只要是引用表的地方都可以追加定义。在指定的表后面追加的动态参数会自动追加到原表定义中，是不是很方便呢：）

由于可能对查询结果有影响，动态参数功能默认是关闭的，使用下面的方式开启该功能：

// instantiate table environment

TableEnvironment tEnv = ...

// access flink configuration

Configuration configuration = tEnv.getConfig().getConfiguration();

// set low-level key-value options

configuration.setString("table.dynamic-table-options.enabled", "true");

细节参见：https://ci.apache.org/projects/flink/flink-docs-master/dev/table/sql/hints.html

SQL API 改进

随着 Flink SQL 支持的语句越来越丰富，老的 API 容易引起一些困惑：

原先的 sqlUpdate() 方法传递 DDL 语句会立即执行，而 INSERT INTO 语句在调用 execute 方法时才会执行；
Table 程序的执行入口不够清晰，像 TableEnvironment.execute() 和 StreamExecutionEnvironment.execute() 都可以触发 table 程序执行；
execute 方法没有返回值。像 SHOW TABLES 这样的语句没有很好地方式返回结果。另外，sqlUpdate 方法加入了越来越多的语句导致接口定义不清晰，sqlUpdate 可以执行 SHOW TABLES 就是一个反例；
在 Blink planner 一直提供多 sink 优化执行的能力，但是在 API 层没有体现出来。

1.11 重新梳理了 TableEnv 上的 sql 相关接口，提供了更清晰的执行语义，同时执行任意 sql 语句现在都有返回值，用户可以通过新的 API 灵活的组织多行 sql 语句一起执行。

更清晰的执行语义

新的接口 TableEnvironment#executeSql 统一返回抽象 TableResult，用户可以迭代 TableResult 拿到执行结果。根据执行语句的不同，返回结果的数据结构也有变化，比如 SELECT 语句会返回查询结果，而 INSERT 语句会异步提交作业到集群。

组织多条语句一起执行

新的接口 TableEnvironment#createStatementSet 允许用户添加多条 INSERT 语句并一起执行，在多 sink 场景，Blink planner 会针对性地对执行计划做优化。

新旧 API 对比

一张表格感受新老 API 的变化：

sqlUpdate vs executeSql

execute vs createStatementSet

详情参见：https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=134745878

Hive 语法兼容加强

从 1.11 开始，Flink SQL 将 Hive parser 模块独立出来，用以兼容 Hive 的语法，目前 DDL 层面，DB、Table、View、Function 相关的语法均已支持。搭配 HiveCatalog，Hive 的同学可以直接使用 Hive 的语法来进行相关的操作。

在使用 hive 语句之前需要设置正确的 Dialect:

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

TableEnvironment tableEnv = TableEnvironment.create(settings);

// to use hive dialect

tableEnv.getConfig().setSqlDialect(SqlDialect.HIVE);

// use the hive catalog

tableEnv.registerCatalog(hiveCatalog.getName(), hiveCatalog);

tableEnv.useCatalog(hiveCatalog.getName());

之后我们便可以使用 Hive 的语法来执行一些 DDL，例如最常见的建表操作：

create external table tbl1 (

d decimal(10,0),

ts timestamp)

partitioned by (p string)

location '%s'

tblproperties('k1'='v1');

create table tbl2 (s struct<ts:timestamp,bin:binary>) stored as orc;

create table tbl3 (

m map<timestamp,binary>

)

partitioned by (p1 bigint, p2 tinyint)

row format serde 'org.apache.hadoop.hive.serde2.lazybinary.LazyBinarySerDe';

create table tbl4 (

x int,

y smallint)

row format delimited fields terminated by '|' lines terminated by '\n';

对于 DQL 的 Hive 语法兼容已经在规划中，1.12 版本会兼容更多 query 语法 ~

详情参见：https://ci.apache.org/projects/flink/flink-docs-master/dev/table/hive/hive_dialect.html

更简洁的 connector 属性

1.11 重新规范了 connector 的属性定义，新的属性 key 更加直观简洁，和原有的属性 key 相比主要做了如下改动：

使用 connector 作为 connector 的类型 key，connector 版本信息直接放到 value 中，比如 0.11 的 kafka 为 kafka-0.11 ；
去掉了其余属性中多余的 connector 前缀；
使用 scan 和 sink 前缀标记 source 和 sink 专有属性；
format.type 精简为 format ，同时 format 自身属性使用 format 的值作为前缀，比如 csv format 的自身属性使用 csv 统一作前缀。

例如，1.11 Kafka 表的定义如下：

CREATE TABLE kafkaTable (

user_id BIGINT,

item_id BIGINT,

category_id BIGINT,

behavior STRING,

ts TIMESTAMP(3)

) WITH (

'connector' = 'kafka',

'topic' = 'user_behavior',

'properties.bootstrap.servers' = 'localhost:9092',

'properties.group.id' = 'testGroup',

'format' = 'csv',

'scan.startup.mode' = 'earliest-offset'

)

详情参见：https://cwiki.apache.org/confluence/display/FLINK/FLIP-122%3A+New+Connector+Property+Keys+for+New+Factory

JDBC catalog

在之前的版本中，用户只能通过显示建表的方式创建关系型数据库的镜像表。用户需要手动追踪 Flink SQL 的表 schema 和数据库的 schema 变更。在 1.11，Flink SQL 提供了一个 JDBC catalog 接口对接各种外部的数据库系统，例如 Postgres、MySQL、MariaDB、AWS Aurora、etc。

当前 Flink 内置了 Postgres 的 catalog 实现，使用下面的代码配置 JDBC catalog：

CREATE CATALOG mypg WITH(

'type' = 'jdbc',

'default-database' = '...',

'username' = '...',

'password' = '...',

'base-url' = '...'

);

USE CATALOG mypg;

用户也可以实现 JDBCCatalog 接口定制其他数据库的 catalog ~

详情参见：https://ci.apache.org/projects/flink/flink-docs-master/dev/table/connectors/jdbc.html#postgres-database-as-a-catalog

Python UDF 增强

1.11 版本的 py-flink 在 python UDF 方面提供了很多增强，包括 DDL 的定义方式、支持了标量的向量化 python UDF，支持全套的 python UDF metrics 定义，以及在 SQL-CLI 中定义 python UDF。

DDL 定义 python UDF

1.10.0 版本引入了对 python UDF 的支持。但是仅仅支持 python table api 的方式。1.11 提供了 SQL DDL 的方式定义 python UDF, 用户可以在 Java/Scala table API 以及 SQL-CLI 场景下使用。

例如，现在用户可以使用如下方式定义 Java table API 程序使用 python UDF：

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

BatchTableEnvironment tEnv = BatchTableEnvironment.create(env);

tEnv.getConfig().getConfiguration().setString("python.files", "/home/my/test1.py");

tEnv.getConfig().getConfiguration().setString("python.client.executable", "python3");

tEnv.sqlUpdate("create temporary system function func1 as 'test1.func1' language python");

Table table = tEnv.fromDataSet(env.fromElements("1", "2", "3")).as("str").select("func1(str)");

tEnv.toDataSet(table, String.class).collect();

向量化支持

向量化 Python UDF 相较于普通函数大大提升了性能。用户可以使用流行的 python 库例如 Pandas、Numpy 来实现向量化的 python UDF。用户只需在装饰器 udf 中添加额外的参数 udf_type="pandas" 即可。

例如，下面的样例展示了如何定义向量化的 Python 标量函数以及在 python table api 中的应用：

@udf(input_types=[DataTypes.BIGINT(), DataTypes.BIGINT()], result_type=DataTypes.BIGINT(), udf_type="pandas")

def add(i, j):

return i + j

table_env = BatchTableEnvironment.create(env)

# register the vectorized Python scalar function

table_env.register_function("add", add)

# use the vectorized Python scalar function in Python Table API

my_table.select("add(bigint, bigint)")

# use the vectorized Python scalar function in SQL API

table_env.sql_query("SELECT add(bigint, bigint) FROM MyTable")

详情参见：https://ci.apache.org/projects/flink/flink-docs-master/dev/table/python/vectorized_python_udfs.html

另外，1.11 对 python UDF 的 metrics 做了全面支持，现在用户可以在 UDF 中方便地定义各种类型的 metrics，由于篇幅关系，这里不作详细描述，见 python UDF metrics。

详情参见：https://ci.apache.org/projects/flink/flink-docs-master/dev/table/python/metrics.html

展望后续

在后续版本，易用性仍然是 Flink SQL 的核心主题，比如 schema 的易用性增强，Descriptor API 简化以及更丰富的流 DDL 将会是努力的方向，让我们拭目以待 ~

作者丨陈玉兆（玉兆）

来源丨Flink 中文社区（ID：gh_5efd76d10a8d）

dbaplus社群欢迎广大技术人员投稿，投稿邮箱：editor@dbaplus.cn

Flink 1.11新版本解读：更易用的流批一体SQL引擎

降本的Kubernetes何时成了“成本刺客”？

YouTube如何利用MySQL支撑24.9亿用户？

人肉运维100次后，年底出了P0级故障……

关于国产数据库我不得不说

分库分表，可能真的要退出历史舞台了！