Flink 1.10 SQL、HiveCatalog 與事件時間整合示例

摘要：flink-sql-connector-kafka-0.11_${scala.bin.version}。flink-connector-hive_${scala.bin.version}。

Flink 1.10 與 1.9 相比又是個創新版本，在我們感興趣的很多方面都有改進，特別是 Flink SQL。本文用根據埋點日誌計算 PV、UV 的簡單示例來體驗 Flink 1.10 的兩個重要新特性：

一是 SQL DDL 對事件時間的支持；
二是 Hive Metastore 作爲 Flink 的元數據存儲（即 HiveCatalog）。

這兩點將會爲我們構建實時數倉提供很大的便利。

添加依賴項

示例採用 Hive 版本爲 1.1.0，Kafka 版本爲 0.11.0.2。

要使 Flink 與 Hive 集成以使用 HiveCatalog，需要先將以下 JAR 包放在 ${FLINK_HOME}/lib 目錄下。

flink-connector-hive_2.11-1.10.0.jar
flink-shaded-hadoop-2-uber-2.6.5-8.0.jar
hive-metastore-1.1.0.jar
hive-exec-1.1.0.jar
libfb303-0.9.2.jar

後三個 JAR 包都是 Hive 自帶的，可以在 ${HIVE_HOME}/lib 目錄下找到。前兩個可以通過阿里雲 Maven 搜索 GAV 找到並手動下載（groupId 都是org.apache.flink）。

再在 pom.xml 內添加相關的 Maven 依賴。

Maven 下載：

https://maven.aliyun.com/mvn/search

<properties>
    <scala.bin.version>2.11</scala.bin.version>
    <flink.version>1.10.0</flink.version>
    <hive.version>1.1.0</hive.version>
  </properties>

  <dependencies>
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-table-api-scala_${scala.bin.version}</artifactId>
      <version>${flink.version}</version>
    </dependency>
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-table-api-scala-bridge_${scala.bin.version}</artifactId>
      <version>${flink.version}</version>
    </dependency>
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-table-planner-blink_${scala.bin.version}</artifactId>
      <version>${flink.version}</version>
    </dependency>
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-sql-connector-kafka-0.11_${scala.bin.version}</artifactId>
      <version>${flink.version}</version>
    </dependency>
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-connector-hive_${scala.bin.version}</artifactId>
      <version>${flink.version}</version>
    </dependency>
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-json</artifactId>
      <version>${flink.version}</version>
    </dependency>
    <dependency>
      <groupId>org.apache.hive</groupId>
      <artifactId>hive-exec</artifactId>
      <version>${hive.version}</version>
    </dependency>
  </dependencies>

最後，找到 Hive 的配置文件 hive-site.xml，準備工作就完成了。

註冊 HiveCatalog、創建數據庫

不多廢話了，直接上代碼，簡潔易懂。

val streamEnv = StreamExecutionEnvironment.getExecutionEnvironment
    streamEnv.setParallelism(5)
    streamEnv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val tableEnvSettings = EnvironmentSettings.newInstance()
        .useBlinkPlanner()
        .inStreamingMode()
        .build()
    val tableEnv = StreamTableEnvironment.create(streamEnv, tableEnvSettings)

    val catalog = new HiveCatalog(
      "rtdw",                   // catalog name
      "default",                // default database
      "/Users/lmagic/develop",  // Hive config (hive-site.xml) directory
      "1.1.0"                   // Hive version
    )
    tableEnv.registerCatalog("rtdw", catalog)
    tableEnv.useCatalog("rtdw")

    val createDbSql = "CREATE DATABASE IF NOT EXISTS rtdw.ods"
    tableEnv.sqlUpdate(createDbSql)

創建 Kafka 流表並指定事件時間

我們的埋點日誌存儲在指定的 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，圖中未示出。

開窗計算 PV、UV

用30秒的滾動窗口，按事件類型來分組，查詢語句如下。

SELECT eventType,
TUMBLE_START(eventTime, INTERVAL '30' SECOND) AS windowStart,
TUMBLE_END(eventTime, INTERVAL '30' SECOND) AS windowEnd,
COUNT(userId) AS pv,
COUNT(DISTINCT userId) AS uv
FROM rtdw.ods.streaming_user_active_log
WHERE platform = 'xyz'
GROUP BY eventType, TUMBLE(eventTime, INTERVAL '30' SECOND)

關於窗口在 SQL 裏的表達方式請參見官方文檔。1.10 版本 SQL 的官方文檔寫的還是比較可以的。

SQL 文檔：

https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/table/sql/queries.html#group-windows

懶得再輸出到一個結果表了，直接轉換成流打到屏幕上。

    val queryActiveSql =
      """
        |......
        |......
      """.stripMargin
    val result = tableEnv.sqlQuery(queryActiveSql)

    result
        .toAppendStream[Row]
        .print()
        .setParallelism(1)