?實時同步MySQL數據到Hive

案例：將mysql中新增的數據實時同步到Hive中。

以上案例需要用到的處理器有：“CaptureChangeMySQL”、“RouteOnAttribute”、“EvaluateJsonPath”、“ReplaceText”、“PutHiveQL”。

首先通過“CaptureChangeMySQL”讀取MySQL中數據的變化（需要開啟MySQL binlog日志），將Binlog中變化的數據同步到“RouteOnAttribute”處理器，通過此處理器獲取上游數據屬性，獲取對應binlog操作類型，再將想要處理的數據路由到“EvaluateJsonPath”處理器，該處理器可以將json格式的binlog數據解析，通過自定義json 表達式獲取json數據中的屬性放入FlowFile屬性,將FlowFile通過“ReplaceText”處理器獲取上游FowFile屬性，動態拼接sql替換所有的FlowFile內容，將拼接好的sql組成FlowFile路由到“PutHiveQL”將數據寫入到Hive表。

(資料圖片)

一、開啟MySQL的binlog日志

mysql-binlog是MySQL數據庫的二進制日志，記錄了所有的DDL和DML(除了數據查詢語句)語句信息。一般來說開啟二進制日志大概會有1%的性能損耗。這里需要開啟MySQL的binlog日志方便后期使用“CaptureChangeMySQL”處理器來獲取MySQL中的CDC事件。MySQL的版本最好是5.7版本之上。

1、登錄mysql查看MySQL是否開啟binlog日志

[root@node2 ~]# mysql -u root -p123456mysql> show variables like "log_%";

2 、開啟mysql binlog日志

在/etc/my.cnf文件中[mysqld]下寫入以下內容：

[mysqld]#隨機指定一個不能和其他集群中機器重名的字符串server-id=123#配置binlog日志目錄，配置后會自動開啟binlog日志，并寫入該目錄log-bin=/var/lib/mysql/mysql-bin

3、重啟mysql 服務，重新查看binlog日志情況

[root@node2 ~]# service mysqld restart[root@node2 ~]# mysql -u root -p123456mysql> show variables like "log_%";

二、???????配置“CaptureChangeMySQL”處理器

“CaptureChangeMySQL”主要是從MySQL數據庫捕獲CDC(Change Data Capture)事件。CDC事件包括INSERT，UPDATE，DELETE操作，事件按操作發生時的順序輸出為單獨的FlowFile文件。

關于“CaptureChangeMySQL”處理器的“Properties”主要配置的說明如下：

配置步驟如下：

1、創建“CaptureChangeMySQL”處理器

2、配置“DistributeMapCacheServer”控制服務

監控mysql變化需要設置“DistributedMapCacheClient”控制服務，其對應的Server中存儲處理器所需的各種表、列等信息，所以這里需要首先配置“DistributeMapCacheServer”控制服務。

?

?

3、配置“SCHEDULING”

由于這里使用“CaptureChangeMySQL”處理器監控“MySQL”中的數據，所以設置調度訪問周期為“10s”，防止一直監聽MySQL binlog數據，帶來性能消耗。

?

4、配置“PROPERTIES”

在“CaptureChangeMySQL”處理器中配置“PROPERTIES”，配置如下：

注意：這里需要在每臺NiFi節點上創建對應目錄，上傳mysql驅動包。

“PROPERTIES”配置如下：

此外，在“PROPERTIES”中還需要配置“Distributed Map Cache Client”控制服務，來讀取“DistributeMapCacheServer”控制服務中的緩存數據：

?

另外，這里我們只是監控表“test2”對應的CDC事件，這里設置匹配表名為“test2”,最終“PROPERTIES”的配置如下：

注意：以上“Table Name Pattern”這里配置對應的Value值為：test2，也可以不配置，不配置會監控所有MySQL表的變化對應的binlog事件。當后面向Hive表中插入新增和更新數據時，對應MySQL中的元數據表也會變化，也會監控到對應的binlog事件。為了避免后期出現監控到其他表的binlog日志，這里建議配置上“test2”。

5、啟動MySQL，創建表“test2”測試“CaptureChangeMySQL”處理器

登錄mysql ,使用“mynifi”庫，創建表“test2”。暫時設置“CaptureChangeMySQL”處理器“success”事件自動終止并啟動，向表中插入對應的數據查看“CaptureChangeMySQL”處理器能否正常監控事件。

在mysql中創建對應的表：

use mynifi;create table test2 (id int,name varchar(255),age int);

啟動“CaptureChangeMySQL”處理器：

向表“test2”中插入以下數據：

insert into test2 values (1,"zs",18);update test2 set name = "ls" where id = 1;delete from test2 where id = 1;

可以在“CaptureChangeMySQL”處理器中右鍵“View data provenance”查看捕獲到的“insert”、“update”、“delete”事件：

注意問題：在配置好“CaptureChangeMySQL”處理器啟動后，當MySQL中有數據插入、修改、刪除時當前處理器會讀取MySql binlog日志，并在當前處理器中記錄讀取binlog的位置狀態。正常來說這里關閉“CaptureChangeMySQL”處理器后再次啟動，會接著保存的binlog位置繼續讀取（可以參照“PROPERTIES”屬性中“Retrieve All Records”配置說明），但是經過測試，此NiFi版本出現以下錯誤（無效的binlog位置，目測是一個版本bug錯誤）：

所以在之后的測試中，我們可以將“CaptureChangeMysql”處理器讀取binlog的狀態清空，然后再次啟動即可，這里會重復讀取MySQL之前已經檢測到的新增、修改、刪除數據。

清空“CaptureChangeMysql”讀取binlog狀態：

三、??????????????配置“RouteOnAttribute”處理器

“RouteOnAttribute”是根據FlowFile的屬性使用屬性表達式進行數據路由。

關于“RouteOnAttribute”處理器的“Properties”主要配置的說明如下：

注意：該處理器允許用戶自定義屬性并指定該屬性的匹配表達式。屬性與動態屬性指定的屬性表達式相匹配的FileFlow，映射到動態屬性上。

配置如下：

1、創建“RouteOnAttribute”處理器

2、配置“PROPERTIES”自定義屬性

注意：以上自定義的屬性中update、insert、delete對應的json 表達式寫法為：${cdc.event.type:equals("delete")}，代表匹配對應類型的FlowFile,“cdc.event.type”是上游FlowFile中的屬性，“equales”是對應的方法，“delete”使用單引號引起，表示匹配的CDC事件。

3、連接“CaptureChangeMySQL”處理器與“RouteOnAttribute”處理器

四、配置“EvaluatejsonPath”處理器

“EvaluatejsonPath”處理器將根據上游“RouteOnAttribute”匹配的事件將內容映射成FlowFile屬性，方便后期拼接SQL獲取數據，上游匹配到的FlowFile中的數據格式為：

EvaluatejsonPath”處理器配置如下：

1、配置“EvaluatejsonPath”的“PROPERTIES”屬性

2、連接“RouteOnAttribute”處理器和“EvaluatejsonPath”處理器

連接關系中，我們這里只關注“insert”和“update”的數據，后期獲取對應的屬性將插入和更新的數據插入到Hive表中，對于“delete”的數據可以路由到其他關系中，例如需要將刪除數據插入到另外的Hive表中，可以再設置個分支處理。這里我們將“delete”和“failure”的數據設置自動終止關系。

設置“RouteOnAttribute”處理器其他匹配路由關系為自動終止：

五、??????????????配置“ReplaceText”處理器

“ReplaceText”處理器可以獲取“EvaluatejsonPath”轉換后FlowFile中的屬性來替換原有數據組成一個“insert into ... values (... ...)”語句，方便后續將數據插入到Hive中。“ReplaceText”處理器的配置如下：

1、配置“RelaceText”處理器“PROPERTIES”屬性

在“Replacement Value”中配置“insert into ${tablename} values (${id},"${name}",${age})”

注意：

以上獲取的tablename名稱為“test2”，后面這個sql是要將數據插入到Hive中的，所以這里在Hive中也應該創建“test2”的表名稱，或者將表名稱寫成固定表，后期在Hive中創建對應的表即可。

另外，需要注意${name}在插入Hive中時對應的列為字符串，這里需要加上單引號。

2、連接“EvaluatejsonPath”處理器與“ReplaceText”處理器

配置“EvaluatjsonPath”處理器“failure”和“unmatch”路由關系為自動終止。

六、??????????????配置Hive 支持HiveServer2

訪問Hive有兩種方式：HiveServer2和Hive Client，Hive Client需要Hive和Hadoop的jar包，配置環境。HiveServer2使得連接Hive的Client從Yarn和HDFS集群中獨立出來，不需要每個幾點都配置Hive和Hadoop的jar包和一系列環境。

NiFi連接Hive就是使用了HiveServer2方式連接，所以這里需要配置HiveServer2。

配置HiveServer2步驟如下：

1、在Hive服務端配置hive-site.xml

#在Hive 服務端 $HIVE_HOME/etc/hive-site.xml中配置： hive.server2.thrift.port 10000hive.server2.thrift.bind.host192.168.179.4

2、在每臺Hadoop 節點配置core-site.xml

     hadoop.proxyuser.root.hosts     *       hadoop.proxyuser.root.groups        * 

3、重啟HDFS ，Hive ，在Hive服務端啟動Metastore和HiveServer2服務

nohup hive --service metastore >> ./nohup.out 2>&1 &nohup hive --service hiveserver2 >> ./nohup.out 2>&1 &

4、在客戶端通過beeline連接Hive

[root@node3 test]# beelinebeeline> !connect jdbc:hive2://node1:10000 rootEnter password for jdbc:hive2://node1:10000: 沒有密碼直接跳過即可0: jdbc:hive2://node1:10000> show tables;+------------------------------------+|              tab_name              |+------------------------------------+| personinfo                         || test2                              |+------------------------------------+

以上配置完成后，還需要將配置好的core-site.xml文件發送到各個NiFi節點對應的路徑/root/test下替換原有的core-site.xml文件。之后重啟NiFi集群，各個NiFi節點上執行命令：

service nifi restart

七、配置“PutHiveQL”處理器

“PutHiveQL”主要執行HiveQL的DDL/DML命令，傳入給該處理器的FlowFile內容是要執行的HiveQL命令。HiveQL命令可以使用“？”來指定參數，這種情況下，參數必須存在于FlowFile的屬性中，命名約定為hiveql.args.N.type和hiveql.args.N.value，其中N為正整數。

關于“PutHiveQL”處理器的“Properties”主要配置的說明如下：

“PutHiveQL”處理器的配置如下：

1、創建“PutHiveQL”處理器

?

2、 配置“PROPERTIES”

?

點擊之后，配置“HiveConnectionPool”控制服務：

注意以上需要配置：

配置完成后，需要啟用對應的“HiveConnectionPool”控制服務：

最終配置“PROPERTIES”為：

3、連接“ReplaceText”處理器與“PutHiveQL”處理器并設置關系

?

設置“ReplaceText”處理器“failure”路由關系為自動終止：

設置“PutHiveQL”處理器路由關系為自動終止：

?

八、??????????????運行測試

1、在Hive中創建表“test2”

動HDFS,啟動Hive服務端和客戶端，創建表“test2”

create table test2 (id int,name string,age int )row format delimited fields terminated by "\t";

2、啟動NiFi處理數據流程,向MySQL中寫入數據，查看Hive中表數據

首先清空“CaptureChangeMySQL”處理器的狀態，單獨啟動“CaptureChangeMySQL”處理器，清空重新消費的數據（以上主要就是避免此版本NiFi bug問題），啟動當前案例中其他NiFi處理器。

然后向MySQL中插入以下數據：

insert into test2 values (1,"zs",18);update test2 set name = "ls" where id = 1;delete from test2 where id = 1;

NiFi頁面：

Hive表test2中的結果：