Flink SQL Client綜合實戰

在《Flink SQL Client初探》一文中，咱們體驗了Flink SQL Client的基本功能，今天來經過實戰更深刻學習和體驗Flink SQL；java

實戰內容

本次實戰主要是經過Flink SQL Client消費kafka的實時消息，再用各類SQL操做對數據進行查詢統計，內容彙總以下：mysql

DDL建立Kafka表
窗口統計；
數據寫入ElasticSearch
聯表操做

版本信息

Flink：1.10.0
Flink所在操做系統：CentOS Linux release 7.7.1908
JDK：1.8.0_211
Kafka：2.4.0（scala：2.12）
Mysql：5.7.29

數據源準備

本次實戰用的數據，來源是阿里雲天池公開數據集的一份淘寶用戶行爲數據集，獲取方式請參考《準備數據集用於flink學習》
獲取到數據集文件後轉成kafka消息發出，這樣咱們使用Flink SQL時就按照實時消費kafka消息的方式來操做，具體的操做方式請參考《將CSV的數據發送到kafka》
上述操做完成後，一百零四萬條淘寶用戶行爲數據就會經過kafka消息順序發出，我們的實戰就有不間斷實時數據可用了，消息內容以下：

{"user_id":1004080,"item_id":2258662,"category_id":79451,"behavior":"pv","ts":"2017-11-24T23:47:47Z"}
{"user_id":100814,"item_id":5071478,"category_id":1107469,"behavior":"pv","ts":"2017-11-24T23:47:47Z"}
{"user_id":114321,"item_id":4306269,"category_id":4756105,"behavior":"pv","ts":"2017-11-24T23:47:48Z"}

上述消息中每一個字段的含義以下表：

列名稱	說明
用戶ID	整數類型，序列化後的用戶ID
商品ID	整數類型，序列化後的商品ID
商品類目ID	整數類型，序列化後的商品所屬類目ID
行爲類型	字符串，枚舉類型，包括(‘pv’, ‘buy’, ‘cart’, ‘fav’)
時間戳	行爲發生的時間戳
時間字符串	根據時間戳字段生成的時間字符串

jar準備

實戰過程當中要用到下面這五個jar文件：git

flink-jdbc_2.11-1.10.0.jar
flink-json-1.10.0.jar
flink-sql-connector-elasticsearch6_2.11-1.10.0.jar
flink-sql-connector-kafka_2.11-1.10.0.jar
mysql-connector-java-5.1.48.jar

我已將這些文件打包上傳到GitHub，下載地址：https://raw.githubusercontent.com/zq2599/blog_demos/master/files/sql_lib.zip程序員

請在flink安裝目錄下新建文件夾sql_lib，而後將這五個jar文件放進去；github

Elasticsearch準備

若是您裝了docker和docker-compose，那麼下面的命令能夠快速部署elasticsearch和head工具：sql

wget https://raw.githubusercontent.com/zq2599/blog_demos/master/elasticsearch_docker_compose/docker-compose.yml && \
docker-compose up -d

準備完畢，開始操做吧；docker

DDL建立Kafka表

進入flink目錄，啓動flink：bin/start-cluster.sh
啓動Flink SQL Client：bin/sql-client.sh embedded -l sql_lib
啓動成功顯示以下：
執行如下命令便可建立kafka表，請按照本身的信息調整參數：

CREATE TABLE user_behavior (
    user_id BIGINT,
    item_id BIGINT,
    category_id BIGINT,
    behavior STRING,
    ts TIMESTAMP(3),
    proctime as PROCTIME(),   -- 處理時間列
    WATERMARK FOR ts as ts - INTERVAL '5' SECOND  -- 在ts上定義watermark，ts成爲事件時間列
) WITH (
    'connector.type' = 'kafka',  -- kafka connector
    'connector.version' = 'universal',  -- universal 支持 0.11 以上的版本
    'connector.topic' = 'user_behavior',  -- kafka topic
    'connector.startup-mode' = 'earliest-offset',  -- 從起始 offset 開始讀取
    'connector.properties.zookeeper.connect' = '192.168.50.43:2181',  -- zk 地址
    'connector.properties.bootstrap.servers' = '192.168.50.43:9092',  -- broker 地址
    'format.type' = 'json'  -- 數據源格式爲 json
);

執行SELECT * FROM user_behavior;看看原始數據，若是消息正常應該和下圖相似：

窗口統計

下面的SQL是以每十分鐘爲窗口，統計每一個窗口內的總瀏覽數，TUMBLE_START返回的數據格式是timestamp，這裏再調用DATE_FORMAT函數將其格式化成了字符串：

SELECT DATE_FORMAT(TUMBLE_START(ts, INTERVAL '10' MINUTE), 'yyyy-MM-dd hh:mm:ss'), 
DATE_FORMAT(TUMBLE_END(ts, INTERVAL '10' MINUTE), 'yyyy-MM-dd hh:mm:ss'), 
COUNT(*)
FROM user_behavior
WHERE behavior = 'pv'
GROUP BY TUMBLE(ts, INTERVAL '10' MINUTE);

獲得數據以下所示：

數據寫入ElasticSearch

確保elasticsearch已部署好；
執行如下語句便可建立es表，請按照您本身的es信息調整下面的參數：

CREATE TABLE pv_per_minute ( 
    start_time STRING,
    end_time STRING,
    pv_cnt BIGINT
) WITH (
    'connector.type' = 'elasticsearch', -- 類型
    'connector.version' = '6',  -- elasticsearch版本
    'connector.hosts' = 'http://192.168.133.173:9200',  -- elasticsearch地址
    'connector.index' = 'pv_per_minute',  -- 索引名，至關於數據庫表名
    'connector.document-type' = 'user_behavior', -- type，至關於數據庫庫名
    'connector.bulk-flush.max-actions' = '1',  -- 每條數據都刷新
    'format.type' = 'json',  -- 輸出數據格式json
    'update-mode' = 'append'
);

執行如下語句，就會將每分鐘的pv總數寫入es的pv_per_minute索引：

INSERT INTO pv_per_minute
SELECT DATE_FORMAT(TUMBLE_START(ts, INTERVAL '1' MINUTE), 'yyyy-MM-dd hh:mm:ss') AS start_time, 
DATE_FORMAT(TUMBLE_END(ts, INTERVAL '1' MINUTE), 'yyyy-MM-dd hh:mm:ss') AS end_time, 
COUNT(*) AS pv_cnt
FROM user_behavior
WHERE behavior = 'pv'
GROUP BY TUMBLE(ts, INTERVAL '1' MINUTE);

用es-head查看，發現數據已成功寫入：

聯表操做

當前user_behavior表的category_id表示商品類目，例如11120表示計算機書籍，61626表示牛仔褲，本次實戰的數據集中，這樣的類目共有五千多種；
若是咱們將這五千多種類目分紅6個大類，例如11120屬於教育類，61626屬於服裝類，那麼應該有個大類和類目的關係表；
這個大類和類目的關係表在MySQL建立，表名叫category_info，建表語句以下：

CREATE TABLE `category_info`(
   `id` int(11) unsigned NOT NULL AUTO_INCREMENT,
   `parent_id` bigint ,
   `category_id` bigint ,
   PRIMARY KEY ( `id` )
) ENGINE=InnoDB AUTO_INCREMENT=5 DEFAULT CHARSET=utf8 COLLATE=utf8_bin;

表category_info全部數據來自對原始數據中category_id字段的提取，而且隨機將它們劃分爲6個大類，該表的數據請在個人GitHub下載：https://raw.githubusercontent.com/zq2599/blog_demos/master/files/category_info.sql
請在MySQL上建表category_info，並將上述數據所有寫進去；
在Flink SQL Client執行如下語句建立這個維表，mysql信息請按您本身配置調整：

CREATE TABLE category_info (
	parent_id BIGINT, -- 商品大類
    category_id BIGINT  -- 商品詳細類目
) WITH (
    'connector.type' = 'jdbc',
    'connector.url' = 'jdbc:mysql://192.168.50.43:3306/flinkdemo',
    'connector.table' = 'category_info',
    'connector.driver' = 'com.mysql.jdbc.Driver',
    'connector.username' = 'root',
    'connector.password' = '123456',
    'connector.lookup.cache.max-rows' = '5000',
    'connector.lookup.cache.ttl' = '10min'
);

嘗試聯表查詢：

SELECT U.user_id, U.item_id, U.behavior, C.parent_id, C.category_id
FROM user_behavior AS U LEFT JOIN category_info FOR SYSTEM_TIME AS OF U.proctime AS C
ON U.category_id = C.category_id;

以下圖，聯表查詢成功，每條記錄都能對應大類：
再試試聯表統計，每一個大類的總瀏覽量：

SELECT C.parent_id, COUNT(*) AS pv_count
FROM user_behavior AS U LEFT JOIN category_info FOR SYSTEM_TIME AS OF U.proctime AS C
ON U.category_id = C.category_id
WHERE behavior = 'pv'
GROUP BY C.parent_id;

以下圖，數據是動態更新的：
執行如下語句，能夠在統計時將大類ID轉成中文名：

SELECT CASE C.parent_id
    WHEN 1 THEN '服飾鞋包'
    WHEN 2 THEN '家裝家飾'
    WHEN 3 THEN '家電'
    WHEN 4 THEN '美妝'
    WHEN 5 THEN '母嬰'
    WHEN 6 THEN '3C數碼'
    ELSE '其餘'
  END AS category_name,
COUNT(*) AS pv_count
FROM user_behavior AS U LEFT JOIN category_info FOR SYSTEM_TIME AS OF U.proctime AS C
ON U.category_id = C.category_id
WHERE behavior = 'pv'
GROUP BY C.parent_id;

效果以下圖：

至此，咱們藉助Flink SQL Client體驗了Flink SQL豐富的功能，若是您也在學習Flink SQL，但願本文能給您一些參考；