Flink 1.10 SQL、HiveCatalog 与事件时间整合示例

我们的埋点日志存储在指定的 Kafka topic 里，为 JSON 格式，简化版 schema 大致如下。

"eventType": "clickBuyNow",

    "userId": "97470180",

    "shareUserId": "",

    "platform": "xyz",

    "columnType": "merchDetail",

    "merchandiseId": "12727495",

    "fromType": "wxapp",

    "siteId": "20392",

    "categoryId": "",

    "ts": 1585136092541

其中 ts 字段就是埋点事件的时间戳（毫秒）。在 Flink 1.9 时代，用 CREATE TABLE 语句创建流表时是无法指定事件时间的，只能默认用处理时间。而在 Flink 1.10 下，可以这样写。

CREATE TABLE rtdw.ods.streaming_user_active_log (

  eventType STRING COMMENT '...',

  userId STRING,

  shareUserId STRING,

  platform STRING,

  columnType STRING,

  merchandiseId STRING,

  fromType STRING,

  siteId STRING,

  categoryId STRING,

  ts BIGINT,

  procTime AS PROCTIME(), -- 处理时间

  eventTime AS TO_TIMESTAMP(FROM_UNIXTIME(ts / 1000, 'yyyy-MM-dd HH:mm:ss')), -- 事件时间

  WATERMARK FOR eventTime AS eventTime - INTERVAL '10' SECOND -- 水印

) WITH (

  'connector.type' = 'kafka',

  'connector.version' = '0.11',

  'connector.topic' = 'ng_log_par_extracted',

  'connector.startup-mode' = 'latest-offset', -- 指定起始offset位置

  'connector.properties.zookeeper.connect' = 'zk109:2181,zk110:2181,zk111:2181',

  'connector.properties.bootstrap.servers' = 'kafka112:9092,kafka113:9092,kafka114:9092',

  'connector.properties.group.id' = 'rtdw_group_test_1',

  'format.type' = 'json',

  'format.derive-schema' = 'true', -- 由表schema自动推导解析JSON

  'update-mode' = 'append'

)

Flink SQL 引入了计算列（computed column）的概念，其语法为 column_name AS computed_column_expression，它的作用是在表中产生数据源 schema 不存在的列，并且可以利用原有的列、各种运算符及内置函数。比如在以上 SQL 语句中，就利用内置的 PROCTIME() 函数生成了处理时间列，并利用原有的 ts 字段与 FROM_UNIXTIME()、TO_TIMESTAMP() 两个时间转换函数生成了事件时间列。

为什么 ts 字段不能直接用作事件时间呢？因为 Flink SQL 规定时间特征必须是 TIMESTAMP(3) 类型，即形如”yyyy-MM-ddTHH:mm:ssZ”格式的字符串，Unix 时间戳自然是不行的，所以要先转换一波。

既然有了事件时间，那么自然要有水印。Flink SQL 引入了 WATERMARK FOR rowtime_column_name AS watermark_strategy_expression 的语法来产生水印，有以下两种通用的做法：

• 单调不减水印（对应 DataStream API 的 AscendingTimestampExtractor）

WATERMARK FOR rowtime_column AS rowtime_column - INTERVAL '0.001' SECOND

• 有界乱序水印（对应 DataStream API 的 BoundedOutOfOrdernessTimestampExtractor）

WATERMARK FOR rowtime_column AS rowtime_column - INTERVAL 'n' TIME_UNIT

上文的 SQL 语句中就是设定了 10 秒的乱序区间。如果看官对水印、AscendingTimestampExtractor 和 BoundedOutOfOrdernessTimestampExtractor 不熟的话，可以参见之前的
这篇
，就能理解为什么会是这样的语法了。

https://www.jianshu.com/p/c612e95a5028

下面来正式建表。

    val createTableSql =

      """

        |上文的SQL语句

        |......

      """.stripMargin

    tableEnv.sqlUpdate(createTableSql)

执行完毕后，我们还可以去到 Hive 执行 DESCRIBE FORMATTED ods.streaming_user_active_log 语句，能够发现该表并没有事实上的列，而所有属性（包括 schema、connector、format 等等）都作为元数据记录在了 Hive Metastore 中。

Flink SQL 创建的表都会带有一个标记属性 is_generic=true，图中未示出。