Flink1.12新特性

SQL Connectors 中的 Metadata 处理

Flink 1.12 中，元数据列是 SQL 标准的扩展，参数中connector和format配置提供的metadata字段。元数据列由METADATA关键字指示。例如，元数据列可用于在 Kafka 记录中读取和写入时间戳，以进行基于时间的操作。连接器和格式文档列出了每个组件的可用元数据字段。但是，在表的架构中声明元数据列是可选的。

//sql CREATE TABLE source( field1 varchar, field2 int, ts TIMESTAMP(3) METADATA FROM'timestamp', partition BIGINT METADATA, offset BIGINT METADATA, proc_time AS PROCTIME() )WITH( 'properties.bootstrap.servers'='110.42.146.82:9092', 'connector'='kafka-x', 'scan.parallelism'='1', 'format'='json', 'topic'='planet_MetaData', 'scan.startup.mode'='latest-offset' ); CREATE TABLE sink_kafka( field1 varchar, field2 int, ts timestamp, partition bigint, offset bigint )WITH( 'properties.bootstrap.servers'='110.42.146.82:9092', 'connector'='kafka-x', 'format'='json', 'topic'='planetTest', 'sink.parallelism'='1' ); create table sink( field1 varchar, field2 int, ts timestamp, partition bigint, offset bigint )with( 'connector'='stream-x', 'print'='true' ); insert intosink select field1, field2, ts, partition, offset from source; insert into sink_kafka select field1, field2, ts, partition, offset from source;

~

上述案例中 Kafka connector available metadata

时态表

Flink1.12中，一张随时间变化的表即可被称为时态表，时态表是一个概念解释，并没有实体。

时态表可以被分为两个实体概念，根据有无ChangeLog来区分定义。

版本表: 如果时态表中的记录可以追踪和并访问它的历史版本，这种表我们称之为版本表，来自数据库的 changelog 可以定义成版本表。

普通表: 如果时态表中的记录仅仅可以追踪并和它的最新版本，这种表我们称之为普通表，来自数据库 或 HBase 的表可以定义成普通表。

声明版本表

在 Flink1.12 中，定义了主键约束(1)和事件时间属性(2)的表就是版本表。

-- 定义一张版本表
CREATE TABLE product_changelog (
  product_id STRING,
  product_name STRING,
  product_price DECIMAL(10, 4),
  update_time TIMESTAMP(3) METADATA FROM 'value.source.timestamp' VIRTUAL,
  PRIMARY KEY(product_id) NOT ENFORCED,      -- (1) 定义主键约束
  WATERMARK FOR update_time AS update_time   -- (2) 通过 watermark 定义事件时间              
) WITH (
  'connector' = 'kafka-x',
  'topic' = 'products',
  'scan.startup.mode' = 'earliest-offset',
  'properties.bootstrap.servers' = 'localhost:9092',
  'format' = 'debezium-json'    -- 时态表限定debezium-json
);

声明版本视图

Flink1.12还支持讲Append-Only流转成changlog流。 先声明一个Append-Only表

-- 定义一张 append-only 表
CREATE TABLE RatesHistory (
    currency_time TIMESTAMP(3),
    currency STRING,
    rate DECIMAL(38, 10),
    WATERMARK FOR currency_time AS currency_time   -- 定义事件时间
) WITH (
  'connector' = 'kafka',
  'topic' = 'rates',
  'scan.startup.mode' = 'earliest-offset',
  'properties.bootstrap.servers' = 'localhost:9092',
  'format' = 'json'                                -- 普通的 append-only 流
)

然后通过去重查询定义版本视图

CREATE VIEW versioned_rates AS              
SELECT currency, rate, currency_time            -- (1) `currency_time` 保留了事件时间
  FROM (
      SELECt *,
      ROW_NUMBER() OVER (PARTITION BY currency  -- (2) `currency` 是去重 query 的 unique key，可以作为主键
         ORDER BY currency_time DESC) AS rowNum 
      FROM RatesHistory )
WHERe rowNum = 1; 

时间区间 Join

如果一个 Join 限定输入时间属性必须在一定的时间限制中（即时间窗口），那么就称之为时间区间 Join。

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

与常规 Join 操作相比，时间区间 Join 只支持带有时间属性的递增表。由于时间属性是单调递增的，Flink 可以从状态中移除过期的数据，而不会影响结果的正确性。

时态表Join

时态表 Join 意味着对任意表（左输入/探针侧）去关联一个时态表（右输入/构建侧）的版本，时态表可以是一张跟踪所有变更记录的表（例如数据库表的 changelog，包含多个表快照），也可以是物化所有变更之后的表（例如数据库表，只有最新表快照）。

Flink 使用了 SQL:2011 标准引入的时态表 Join 语法，时态表 Join 的语法如下:

SELECt [column_list]
FROM table1 [AS ]
[LEFT] JOIN table2 FOR SYSTEM_TIME AS OF table1.{ proctime | rowtime } [AS ]
ON table1.column-name1 = table2.column-name1

基于事件时间的时态 Join

基于事件时间的时态表 join 使用(左侧输入/探针侧) 的 事件时间 去关联(右侧输入/构建侧) 版本表 对应的版本。 基于事件时间的时态表 join 仅支持关版本表或版本视图，版本表或版本视图只能是一个 changelog 流。 但是，Flink 支持将 append-only 流转换成 changelog 流，因此版本表也可以来自一个 append-only 流。 查看声明版本视图 获取更多的信息关于如何声明一张来自 append-only 流的版本表。

将事件时间作为时间属性时，可将 过去 时间属性与时态表一起使用。这允许对两个表中在相同时间点的记录执行 Join 操作。 与基于处理时间的时态 Join 相比，时态表不仅将构建侧记录的最新版本（是否最新由所定义的主键所决定）保存在 state 中，同时也会存储自上一个 watermarks 以来的所有版本（按时间区分）。

例如，在探针侧表新插入一条事件时间时间为 12:30:00 的记录，它将和构建侧表时间点为 12:30:00 的版本根据时态表的概念进行 Join 运算。 因此，新插入的记录仅与时间戳小于等于 12:30:00 的记录进行 Join 计算（由主键决定哪些时间点的数据将参与计算）。

通过定义事件时间，watermarks 允许 Join 运算不断向前滚动，丢弃不再需要的构建侧快照。因为不再需要时间戳更低或相等的记录。

下面的例子展示了订单流关联产品表这个场景举例，orders 表包含了来自 Kafka 的实时订单流，product_changelog 表来自数据库表 products 的 changelog , 产品的价格在数据库表 products 中是随时间实时变化的。

SELECt * FROM product_changelog;

(changelog kind)  update_time product_name price
================= =========== ============ ===== 
+(INSERT)         00:01:00    scooter      11.11
+(INSERT)         00:02:00    basketball   23.11
-(UPDATE_BEFORE)  12:00:00    scooter      11.11
+(UPDATE_AFTER)   12:00:00    scooter      12.99  <= 产品 `scooter` 在 `12:00:00` 时涨价到了 `12.99`
-(UPDATE_BEFORE)  12:00:00    basketball   23.11 
+(UPDATE_AFTER)   12:00:00    basketball   19.99  <= 产品 `basketball` 在 `12:00:00` 时降价到了 `19.99`
-(DELETE)         18:00:00    scooter      12.99  <= 产品 `scooter` 在 `18:00:00` 从数据库表中删除

如果我们想输出 product_changelog 表在 10:00:00 对应的版本，表的内容如下所示：

update_time  product_id product_name price
===========  ========== ============ ===== 
00:01:00     p_001      scooter      11.11
00:02:00     p_002      basketball   23.11

如果我们想输出 product_changelog 表在 13:00:00 对应的版本，表的内容如下所示：

update_time  product_id product_name price
===========  ========== ============ ===== 
12:00:00     p_001      scooter      12.99
12:00:00     p_002      basketball   19.99

通过基于事件时间的时态表 join, 我们可以 join 上版本表中的不同版本：

CREATE TABLE orders (
  order_id STRING,
  product_id STRING,
  order_time TIMESTAMP(3),
  WATERMARK FOR order_time AS order_time  -- defines the necessary event time
) WITH (
...
);

-- 设置会话的时间区间, changelog 里的数据库操作时间是以 epoch 开始的毫秒数存储的，
-- 在从毫秒转化为时间戳时，Flink SQL 会使用会话的时间区间
-- 因此，请根据 changelog 中的数据库操作时间设置合适的时间区间
SET table.local-time-zone=UTC;

-- 声明一张版本表
CREATE TABLE product_changelog (
  product_id STRING,sq
  product_name STRING,
  product_price DECIMAL(10, 4),
  update_time TIMESTAMP(3) METADATA FROM 'value.source.timestamp' VIRTUAL, -- 注意：自动从毫秒数转为时间戳
  PRIMARY KEY(product_id) NOT ENFORCED,      -- (1) defines the primary key constraint
  WATERMARK FOR update_time AS update_time   -- (2) defines the event time by watermark                               
) WITH (
  'connector' = 'kafka',
  'topic' = 'products',
  'scan.startup.mode' = 'earliest-offset',
  'properties.bootstrap.servers' = 'localhost:9092',
  'value.format' = 'debezium-json'
);

-- 基于事件时间的时态表 Join
SELECT
  order_id,
  order_time,
  product_name,
  product_time,
  price
FROM orders AS O
LEFT JOIN product_changelog FOR SYSTEM_TIME AS OF O.order_time AS P
ON O.product_id = P.product_id;

order_id order_time product_name product_time price
======== ========== ============ ============ =====
o_001    00:01:00   scooter      00:01:00     11.11
o_002    00:03:00   basketball   00:02:00     23.11
o_003    12:00:00   scooter      12:00:00     12.99
o_004    12:00:00   basketball   12:00:00     19.99
o_005    18:00:00   NULL         NULL         NULL

基于事件时间的时态表 Join 通常用在通过 changelog 丰富流上数据的场景。

注意: 基于事件时间的时态表 Join 是通过左右两侧的 watermark 触发，请确保为 join 两侧的表设置了合适的 watermark。

注意: 基于事件时间的时态表 Join 的 join key 必须包含时态表的主键，例如：表 product_changelog 的主键 P.product_id 必须包含在 join 条件 O.product_id = P.product_id 中。

基于处理时间的时态 Join

基于处理时间的时态表 join 使用任意表 (左侧输入/探针侧) 的 处理时间 去关联 (右侧输入/构建侧) 普通表的最新版本. 基于处理时间的时态表 join 当前只支持关联普通表或普通视图，且支持普通表或普通视图当前只能是 append-only 流。

如果将处理时间作为时间属性，过去 时间属性将无法与时态表一起使用。根据定义，处理时间总会是当前时间戳。 因此，关联时态表的调用将始终返回底层表的最新已知版本，并且底层表中的任何更新也将立即覆盖当前值。

可以将处理时间的时态 Join 视作简单的 HashMap ，HashMap 中存储来自构建侧的所有记录。 当来自构建侧的新插入的记录与旧值具有相同的 Key 时，旧值会被覆盖。 探针侧的每条记录将总会根据 HashMap 的最新/当前状态来计算。

接下来的示例展示了订单流 Orders 该如何与实时变化的汇率表 Lates 进行基于处理时间的时态 Join 操作，LatestRates 总是表示 HBase 表 Rates 的最新内容。

表 LastestRates 中的数据在时间点 10:15 和 10:30 时是相等的。欧元汇率在时间点 10:52 从 114 变化至 116 。

表 Orders 包含了金额字段 amount 和货币字段 currency 的支付记录数据。例如在 10:15 有一笔金额为 2 欧元的订单记录。

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 中的汇率将以下订单转换，则结果将为：

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

通过时态表 Join，我们可以将上述操作表示为以下 SQL 查询：

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

探针侧表中的每个记录都将与构建侧表的当前版本所关联。 在此示例中，查询使用处理时间作为处理时间，因而新增订单将始终与表 LatestRates 的最新汇率执行 Join 操作。注意，结果对于处理时间来说不是确定的。 基于处理时间的时态表 Join 通常用在通过外部表（例如维度表）丰富流上数据的场景。

与常规 Join 相比，尽管构建侧表的数据发生了变化，但时态表 Join 的变化前结果不会随之变化。

• 对于基于事件时间的时态 Join， join 算子保留 Join 两侧流的状态并通过 watermark 清理。
• 对于基于处理时间的时态 Join， join 算子保留仅保留右侧（构建侧）的状态，且构建侧的状态只包含数据的最新版本，右侧的状态是轻量级的; 对于在运行时有能力查询外部系统的时态表，join 算子还可以优化成不保留任何状态，此时算子是非常轻量级的。

与时间区间 Join 相比，时态表 Join 没有定义决定构建侧记录所属的将被 Join 时间窗口。 探针侧的记录将总是与构建侧在对应处理时间的最新数据执行 Join，因而构建侧的数据可能是任意旧的。

Flink1.12与Flink1.10的差异

Flink1.10维表关联方式

create table sink(
    flag varchar,
    field1 varchar,
    field2 varchar,
    field3 varchar
 )with(
    type='console'
 );
insert into sink
select
    'flink110' as flag
    ,s1.field1 as field1
    ,s1.field2 as field2
    ,s2.field2 as field3  
from source s1  
left join dim as s2 
on s1.field1 = s2.field1;

Flink1.12维表关联方式

create table sink(
    flag varchar,
    field1 varchar,
    field2 varchar,
    field3 varchar
 )with(
    'connector'='stream-x',
    'print'='true'
 );
insert into sink
select
    'flink112' as flag
    ,s1.field1 as field1
    ,s1.field2 as field2
    ,s2.field2 as field3  
from source s1  
left join dim for SYSTEM_TIME AS OF s1.proc_time as s2 
on s1.field1 = s2.field1;

Flink1.10嵌套Json

CREATE TABLE source(
    column.field1 string as field1,
    column.field2 int as field2
 )WITH(
    type ='kafka',
    bootstrapServers ='110.42.146.82:9092',
    offsetReset ='latest',
    topic ='planet',
    charsetName ='utf-8',
    timezone='Asia/Shanghai',
    updateMode ='append',
    enableKeyPartitions ='false',
    topicIsPattern ='false',
    parallelism ='1'
 );


insert into sink
select
    *     
from(
    select
        '110_demo1' as flag
        ,field1 as field1
        ,field2 as field2     
    FROM source );

Flink1.12嵌套Json

CREATE TABLE source(
    `column` ROW,
    proc_time AS PROCTIME() 
 )WITH(
    'properties.bootstrap.servers'='110.42.146.82:9092',
    'connector'='kafka-x',
    'scan.parallelism'='1',
    'format'='json',
    'topic'='planet',
    'scan.startup.mode'='latest-offset'
 );

insert into sink
select
    *     
    ,ROW(field1,field2) as columns 
from(
    select
        '112_demo1' as flag
        ,`column`.field1 as field1
        ,`column`.field2 as field2     
    FROM source );

本文由博客群发一文多发等运营工具平台 OpenWrite 发布