189 8069 5689

​FlinkSQL怎么用

这篇文章将为大家详细讲解有关Flink SQL怎么用，小编觉得挺实用的，因此分享给大家做个参考，希望大家阅读完这篇文章后可以有所收获。

目前成都创新互联已为超过千家的企业提供了网站建设、域名、网站空间、网站托管维护、企业网站设计、静海网站维护等服务，公司将坚持客户导向、应用为本的策略，正道将秉承"和谐、参与、激情"的文化，与客户和合作伙伴齐心协力一起成长，共同发展。

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

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


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：覆盖

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


INCLUDING    
EXCLUDING    
OVERWRITING    
ALL    
✔️    
✔️    
❌    
CONSTRAINTS    
✔️    
✔️    
❌    
PARTITIONS    
✔️    
✔️    
❌    
GENERATED    
✔️    
✔️    
✔️    
OPTIONS    
✔️    
✔️    
✔️    

例如下面的语句:

  
    
  
  
  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 sourceselect id, name from kafka_table1 /*+ OPTIONS('scan.startup.mode'='earliest-offset') */;
-- override table options in joinselect * 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 tableinsert into kafka_table1 /*+ OPTIONS('sink.partitioner'='round-robin') */ select * from kafka_table2;

  
    
  
  
  


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

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

  
    
  
  
  
// instantiate table environmentTableEnvironment tEnv = ...// access flink configurationConfiguration configuration = tEnv.getConfig().getConfiguration();// set low-level key-value optionsconfiguration.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  

Current Interface

New Interface

tEnv.sqlUpdate("CREATE TABLE ...");

TableResult result = tEnv.executeSql("CREATE TABLE ...");

tEnv.sqlUpdate("INSERT INTO ... SELECT ...");

tEnv.execute("test");

TableResult result = tEnv.executeSql("INSERT INTO ... SELECT ...");

execute vs createStatementSet

Current Interface    
New Interface    
tEnv.sqlUpdate("insert into xx ...")
tEnv.sqlUpdate("insert into yy ...")
tEnv.execute("test")
StatementSet ss = tEnv.createStatementSet();
ss.addInsertSql("insert into xx ...");
ss.addInsertSql("insert into yy ...");
TableResult result = ss.execute();

   
tEnv.insertInto("sink1", table1)
tEnv.insertInto("sink2", table2)
tEnv.execute("test")
StatementSet ss = tEnv.createStatementSet();
ss.addInsert("sink1", table1);
ss.addInsert("sink2", table2);
TableResult result = ss.execute()
详情参见：  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 dialecttableEnv.getConfig().setSqlDialect(SqlDialect.HIVE);// use the hive catalogtableEnv.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 ~


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 functiontable_env.register_function("add", add)
# use the vectorized Python scalar function in Python Table APImy_table.select("add(bigint, bigint)")
# use the vectorized Python scalar function in SQL APItable_env.sql_query("SELECT add(bigint, bigint) FROM MyTable")

关于“Flink SQL怎么用”这篇文章就分享到这里了，希望以上内容可以对大家有一定的帮助，使各位可以学到更多知识，如果觉得文章不错，请把它分享出去让更多的人看到。


分享文章：​FlinkSQL怎么用
网站网址：http://cdxtjz.com/article/psdojg.html

其他资讯