数智资源网
首页 首页 大数据 大数据入门 查看内容

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

木马童年 2020-10-14 12:43 15 0

许多的数据科学家、分析师和 BI 用户依赖交互式 SQL 查询分析数据。Flink SQL 是 Flink 的核心模块之一。作为一个分布式的 SQL 查询引擎。Flink SQL 提供了各种异构数据源的联合查询。开发者可以很方便地在一个程序 ...

许多的数据科学家、分析师和 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:覆盖。

     

 

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

   

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

   

例如下面的语句:

   
   

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

 

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

   

execute vs createStatementSet

 

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

 

详情参见: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) stored as orc;

   

create table tbl3 (

 

  m map

 

)

 

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 将会是努力的方向,让我们拭目以待 ~

 

数据科学家 数据源 海量数据
0
为您推荐
Spark原理精讲与推荐系统实践案例,资源教程下载

Spark原理精讲与推荐系统实践案例,资源教程下载

课程名称Spark原理精讲与推荐系统实践案例,资源教程下载课程目录Spark 概述Spark Cor…...

北风网数据结构学习视频,资源教程下载

北风网数据结构学习视频,资源教程下载

课程名称北风网数据结构学习视频,资源教程下载课程目录01第一讲数组02第二讲简单排序…...

大数据时代互联网社交媒体数据的分析与应用课程,资源教程下载

大数据时代互联网社交媒体数据的分析与应用课程,资源

课程名称大数据时代互联网社交媒体数据的分析与应用课程,资源教程下载课程介绍大数据…...

炼数成金完整17周Hadoop完全入门学习视频教程 Hadoop数据分析平台第三版视频教程下载

炼数成金完整17周Hadoop完全入门学习视频教程 Hadoop

课程名称炼数成金完整17周Hadoop完全入门学习视频教程 Hadoop数据分析平台第三版视频…...

数据结构和算法案例-欢乐五子棋,资源教程下载

数据结构和算法案例-欢乐五子棋,资源教程下载

课程名称数据结构和算法案例-欢乐五子棋,资源教程下载课程目录【课时1】创建游戏工程…...

Hadoop2.x大数据平台V3完全入门实战Hadoop平台 Hadopp2.X 大数据平台V3开发视频教程下 ...

Hadoop2.x大数据平台V3完全入门实战Hadoop平台 Hadopp

课程名称Hadoop2.x大数据平台V3完全入门实战Hadoop平台 Hadopp2.X 大数据平台V3开发视…...