?實(shí)時(shí)同步MySQL數(shù)據(jù)到Hive

案例：將mysql中新增的數(shù)據(jù)實(shí)時(shí)同步到Hive中。

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

首先通過(guò)“CaptureChangeMySQL”讀取MySQL中數(shù)據(jù)的變化（需要開(kāi)啟MySQL binlog日志），將Binlog中變化的數(shù)據(jù)同步到“RouteOnAttribute”處理器，通過(guò)此處理器獲取上游數(shù)據(jù)屬性，獲取對(duì)應(yīng)binlog操作類型，再將想要處理的數(shù)據(jù)路由到“EvaluateJsonPath”處理器，該處理器可以將json格式的binlog數(shù)據(jù)解析，通過(guò)自定義json 表達(dá)式獲取json數(shù)據(jù)中的屬性放入FlowFile屬性,將FlowFile通過(guò)“ReplaceText”處理器獲取上游FowFile屬性，動(dòng)態(tài)拼接sql替換所有的FlowFile內(nèi)容，將拼接好的sql組成FlowFile路由到“PutHiveQL”將數(shù)據(jù)寫(xiě)入到Hive表。

(資料圖片)

一、開(kāi)啟MySQL的binlog日志

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

1、登錄mysql查看MySQL是否開(kāi)啟binlog日志

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

2 、開(kāi)啟mysql binlog日志

在/etc/my.cnf文件中[mysqld]下寫(xiě)入以下內(nèi)容：

[mysqld]#隨機(jī)指定一個(gè)不能和其他集群中機(jī)器重名的字符串server-id=123#配置binlog日志目錄，配置后會(huì)自動(dòng)開(kāi)啟binlog日志，并寫(xiě)入該目錄log-bin=/var/lib/mysql/mysql-bin

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

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

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

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

關(guān)于“CaptureChangeMySQL”處理器的“Properties”主要配置的說(shuō)明如下：

配置步驟如下：

1、創(chuàng)建“CaptureChangeMySQL”處理器

2、配置“DistributeMapCacheServer”控制服務(wù)

監(jiān)控mysql變化需要設(shè)置“DistributedMapCacheClient”控制服務(wù)，其對(duì)應(yīng)的Server中存儲(chǔ)處理器所需的各種表、列等信息，所以這里需要首先配置“DistributeMapCacheServer”控制服務(wù)。

?

?

3、配置“SCHEDULING”

由于這里使用“CaptureChangeMySQL”處理器監(jiān)控“MySQL”中的數(shù)據(jù)，所以設(shè)置調(diào)度訪問(wèn)周期為“10s”，防止一直監(jiān)聽(tīng)MySQL binlog數(shù)據(jù)，帶來(lái)性能消耗。

?

4、配置“PROPERTIES”

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

注意：這里需要在每臺(tái)NiFi節(jié)點(diǎn)上創(chuàng)建對(duì)應(yīng)目錄，上傳mysql驅(qū)動(dòng)包。

“PROPERTIES”配置如下：

此外，在“PROPERTIES”中還需要配置“Distributed Map Cache Client”控制服務(wù)，來(lái)讀取“DistributeMapCacheServer”控制服務(wù)中的緩存數(shù)據(jù)：

?

另外，這里我們只是監(jiān)控表“test2”對(duì)應(yīng)的CDC事件，這里設(shè)置匹配表名為“test2”,最終“PROPERTIES”的配置如下：

注意：以上“Table Name Pattern”這里配置對(duì)應(yīng)的Value值為：test2，也可以不配置，不配置會(huì)監(jiān)控所有MySQL表的變化對(duì)應(yīng)的binlog事件。當(dāng)后面向Hive表中插入新增和更新數(shù)據(jù)時(shí)，對(duì)應(yīng)MySQL中的元數(shù)據(jù)表也會(huì)變化，也會(huì)監(jiān)控到對(duì)應(yīng)的binlog事件。為了避免后期出現(xiàn)監(jiān)控到其他表的binlog日志，這里建議配置上“test2”。

5、啟動(dòng)MySQL，創(chuàng)建表“test2”測(cè)試“CaptureChangeMySQL”處理器

登錄mysql ,使用“mynifi”庫(kù)，創(chuàng)建表“test2”。暫時(shí)設(shè)置“CaptureChangeMySQL”處理器“success”事件自動(dòng)終止并啟動(dòng)，向表中插入對(duì)應(yīng)的數(shù)據(jù)查看“CaptureChangeMySQL”處理器能否正常監(jiān)控事件。

在mysql中創(chuàng)建對(duì)應(yīng)的表：

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

啟動(dòng)“CaptureChangeMySQL”處理器：

向表“test2”中插入以下數(shù)據(jù)：

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”事件：

注意問(wèn)題：在配置好“CaptureChangeMySQL”處理器啟動(dòng)后，當(dāng)MySQL中有數(shù)據(jù)插入、修改、刪除時(shí)當(dāng)前處理器會(huì)讀取MySql binlog日志，并在當(dāng)前處理器中記錄讀取binlog的位置狀態(tài)。正常來(lái)說(shuō)這里關(guān)閉“CaptureChangeMySQL”處理器后再次啟動(dòng)，會(huì)接著保存的binlog位置繼續(xù)讀取（可以參照“PROPERTIES”屬性中“Retrieve All Records”配置說(shuō)明），但是經(jīng)過(guò)測(cè)試，此NiFi版本出現(xiàn)以下錯(cuò)誤（無(wú)效的binlog位置，目測(cè)是一個(gè)版本bug錯(cuò)誤）：

所以在之后的測(cè)試中，我們可以將“CaptureChangeMysql”處理器讀取binlog的狀態(tài)清空，然后再次啟動(dòng)即可，這里會(huì)重復(fù)讀取MySQL之前已經(jīng)檢測(cè)到的新增、修改、刪除數(shù)據(jù)。

清空“CaptureChangeMysql”讀取binlog狀態(tài)：

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

“RouteOnAttribute”是根據(jù)FlowFile的屬性使用屬性表達(dá)式進(jìn)行數(shù)據(jù)路由。

關(guān)于“RouteOnAttribute”處理器的“Properties”主要配置的說(shuō)明如下：

注意：該處理器允許用戶自定義屬性并指定該屬性的匹配表達(dá)式。屬性與動(dòng)態(tài)屬性指定的屬性表達(dá)式相匹配的FileFlow，映射到動(dòng)態(tài)屬性上。

配置如下：

1、創(chuàng)建“RouteOnAttribute”處理器

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

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

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

四、配置“EvaluatejsonPath”處理器

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

EvaluatejsonPath”處理器配置如下：

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

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

連接關(guān)系中，我們這里只關(guān)注“insert”和“update”的數(shù)據(jù)，后期獲取對(duì)應(yīng)的屬性將插入和更新的數(shù)據(jù)插入到Hive表中，對(duì)于“delete”的數(shù)據(jù)可以路由到其他關(guān)系中，例如需要將刪除數(shù)據(jù)插入到另外的Hive表中，可以再設(shè)置個(gè)分支處理。這里我們將“delete”和“failure”的數(shù)據(jù)設(shè)置自動(dòng)終止關(guān)系。

設(shè)置“RouteOnAttribute”處理器其他匹配路由關(guān)系為自動(dòng)終止：

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

“ReplaceText”處理器可以獲取“EvaluatejsonPath”轉(zhuǎn)換后FlowFile中的屬性來(lái)替換原有數(shù)據(jù)組成一個(gè)“insert into ... values (... ...)”語(yǔ)句，方便后續(xù)將數(shù)據(jù)插入到Hive中?！癛eplaceText”處理器的配置如下：

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

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

注意：

以上獲取的tablename名稱為“test2”，后面這個(gè)sql是要將數(shù)據(jù)插入到Hive中的，所以這里在Hive中也應(yīng)該創(chuàng)建“test2”的表名稱，或者將表名稱寫(xiě)成固定表，后期在Hive中創(chuàng)建對(duì)應(yīng)的表即可。

另外，需要注意${name}在插入Hive中時(shí)對(duì)應(yīng)的列為字符串，這里需要加上單引號(hào)。

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

配置“EvaluatjsonPath”處理器“failure”和“unmatch”路由關(guān)系為自動(dòng)終止。

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

訪問(wèn)Hive有兩種方式：HiveServer2和Hive Client，Hive Client需要Hive和Hadoop的jar包，配置環(huán)境。HiveServer2使得連接Hive的Client從Yarn和HDFS集群中獨(dú)立出來(lái)，不需要每個(gè)幾點(diǎn)都配置Hive和Hadoop的jar包和一系列環(huán)境。

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

配置HiveServer2步驟如下：

1、在Hive服務(wù)端配置hive-site.xml

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

2、在每臺(tái)Hadoop 節(jié)點(diǎn)配置core-site.xml

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

3、重啟HDFS ，Hive ，在Hive服務(wù)端啟動(dòng)Metastore和HiveServer2服務(wù)

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

4、在客戶端通過(guò)beeline連接Hive

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

以上配置完成后，還需要將配置好的core-site.xml文件發(fā)送到各個(gè)NiFi節(jié)點(diǎn)對(duì)應(yīng)的路徑/root/test下替換原有的core-site.xml文件。之后重啟NiFi集群，各個(gè)NiFi節(jié)點(diǎn)上執(zhí)行命令：

service nifi restart

七、配置“PutHiveQL”處理器

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

關(guān)于“PutHiveQL”處理器的“Properties”主要配置的說(shuō)明如下：

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

1、創(chuàng)建“PutHiveQL”處理器

?

2、 配置“PROPERTIES”

?

點(diǎn)擊之后，配置“HiveConnectionPool”控制服務(wù)：

注意以上需要配置：

配置完成后，需要啟用對(duì)應(yīng)的“HiveConnectionPool”控制服務(wù)：

最終配置“PROPERTIES”為：

3、連接“ReplaceText”處理器與“PutHiveQL”處理器并設(shè)置關(guān)系

?

設(shè)置“ReplaceText”處理器“failure”路由關(guān)系為自動(dòng)終止：

設(shè)置“PutHiveQL”處理器路由關(guān)系為自動(dòng)終止：

?

八、??????????????運(yùn)行測(cè)試

1、在Hive中創(chuàng)建表“test2”

動(dòng)HDFS,啟動(dòng)Hive服務(wù)端和客戶端，創(chuàng)建表“test2”

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

2、啟動(dòng)NiFi處理數(shù)據(jù)流程,向MySQL中寫(xiě)入數(shù)據(jù)，查看Hive中表數(shù)據(jù)

首先清空“CaptureChangeMySQL”處理器的狀態(tài)，單獨(dú)啟動(dòng)“CaptureChangeMySQL”處理器，清空重新消費(fèi)的數(shù)據(jù)（以上主要就是避免此版本NiFi bug問(wèn)題），啟動(dòng)當(dāng)前案例中其他NiFi處理器。

然后向MySQL中插入以下數(shù)據(jù)：

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

NiFi頁(yè)面：

Hive表test2中的結(jié)果：