?離線同步MySQL數據到HDFS

案例：使用NiFi將MySQL中數據導入到HDFS中。

(資料圖片僅供參考)

以上案例用到的處理器有“QueryDatabaseTable”、“ConvertAvroToJSON”、“SplitJson”、“PutHDFS”四個處理器。

一、配置“QueryDatabaseTable”處理器

該處理器主要使用提供的SQL語句或者生成SQL語句來查詢MySQL中的數據，查詢結果轉換成Avro格式。該處理器只能運行在主節點上。

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

配置項	默認值	允許值	描述
Database Connection Pooling Service（數據庫連接池服務）			用于獲得與數據庫的連接的Controller Service。
Database Type（數據庫類型）	Generic		選擇數據庫類型。Generic 通用類型OracleOracle 12+MS SQL 2012+MS SQL 2008MySQLPostgreSQL
Table Name（表名）			查詢數據庫的表名，當使用“Custom Query”時，此為查詢結果的別名，并作為FlowFile中的屬性。
Columns to Return（返回的列）			查詢返回的列，多個列使用逗號分隔。如果列中有特殊名稱需要加引號，則所有列都需要加引號處理。
Additional WHERE clause（where條件）			在構建SQL查詢時添加到WHERE條件中的自定義子句。
Custom Query（自定義SQL查詢）			自定義的SQL語句。該查詢被構建成子查詢，設置后不會從其他屬性構建SQL查詢。自定義SQL不支持Order by查詢。
Maximum-value Columns（最大值列）			指定增量查詢獲取最大值的列，多列使用逗號分開。指定后，這個處理器只能檢索到添加/更新的行。不能設置無法比較大小的列，例如：boolean/bit。如果不指定，則參照表中所有的列來查詢全量數據，這會對性能產生影響。
Max Wait Time（最大超時時間）	0 seconds		SQL查詢最大時長，默認為0沒有限制，設置小于0的時間默認為0。
Fetch Size（拉取數據量）	0		每次從查詢結果中拉取的數據量。
Max Rows Per Flow File（每個FlowFile行數）	0		在一個FlowFile文件中的數據行數。通過這個參數可以將很大的結果集分到多個FlowFile中。默認設置為0，所有結果存入一個FlowFile。
Output Batch Size（數據輸出批次量）	0		輸出的FlowFile批次數據大小，當設置為0代表所有數據輸出到下游關系。如果數據量很大，則有可能下游很久沒有收到數據，如果設置了，則每次達到該數據量就釋放數據，傳輸到下游。
Maximum Number of Fragments（最大片段數）	0		設置返回的最大數據片段數，設置0默認將所有數據片段返回，如果表非常大，設置后可以防止OOM錯誤。
Normalize Table/Column Names（標準表/列名）	false	truefalse	是否將列名中不兼容avro的字符修改為兼容avro的字符。例如，冒號和句點將被更改為下劃線，以構建有效的Avro記錄。
Transaction Isolation Level			設置事務隔離級別。
Use Avro Logical Types(使用Avro邏輯類型)	false	truefalse	是否對DECIMAL/NUMBER, DATE, TIME 和 TIMESTAMP 列使用Avro邏輯類型。
Default Decimal Precision（Decimal數據類型位數）	10		當 DECIMAL/NUMBER 數據類型轉換成Avro類型數據時，指定的數據位數。
Default Decimal Scale（Decimal 數據類型小數位數）	0		當 DECIMAL/NUMBER 數據類型轉換成Avro類型數據時，指定的小數點后的位數。

Generic 通用類型OracleOracle 12+MS SQL 2012+MS SQL 2008MySQLPostgreSQL

Table Name（表名）查詢數據庫的表名，當使用“Custom Query”時，此為查詢結果的別名，并作為FlowFile中的屬性。 Columns to Return （返回的列） 查詢返回的列，多個列使用逗號分隔。如果列中有特殊名稱需要加引號，則所有列都需要加引號處理。 Additional WHERE clause （where條件） 在構建SQL查詢時添加到WHERE條件中的自定義子句。 Custom Query （自定義SQL查詢） 自定義的SQL語句。該查詢被構建成子查詢，設置后不會從其他屬性構建SQL查詢。自定義SQL不支持Order by查詢。 Maximum-value Columns （最大值列） 指定增量查詢獲取最大值的列，多列使用逗號分開。指定后，這個處理器只能檢索到添加/更新的行。不能設置無法比較大小的列，例如：boolean/bit。如果不指定，則參照表中所有的列來查詢全量數據，這會對性能產生影響。 Max Wait Time（最大超時時間）0 seconds SQL查詢最大時長，默認為0沒有限制，設置小于0的時間默認為0。 Fetch Size（拉取數據量）0 每次從查詢結果中拉取的數據量。 Max Rows Per Flow File（每個FlowFile行數）0 在一個FlowFile文件中的數據行數。通過這個參數可以將很大的結果集分到多個FlowFile中。默認設置為0，所有結果存入一個FlowFile。 Output Batch Size（數據輸出批次量）0 輸出的FlowFile批次數據大小，當設置為0代表所有數據輸出到下游關系。如果數據量很大，則有可能下游很久沒有收到數據，如果設置了，則每次達到該數據量就釋放數據，傳輸到下游。 Maximum Number of Fragments（最大片段數）0 設置返回的最大數據片段數，設置0默認將所有數據片段返回，如果表非常大，設置后可以防止OOM錯誤。 Normalize Table/Column Names（標準表/列名）false true false 是否將列名中不兼容avro的字符修改為兼容avro的字符。例如，冒號和句點將被更改為下劃線，以構建有效的Avro記錄。 Transaction Isolation Level 設置事務隔離級別。 Use Avro Logical Types(使用Avro邏輯類型)false true false 是否對DECIMAL/NUMBER, DATE, TIME 和 TIMESTAMP 列使用Avro邏輯類型。 Default Decimal Precision（Decimal數據類型位數）10 當 DECIMAL/NUMBER 數據類型轉換成Avro類型數據時，指定的數據位數。 Default Decimal Scale（Decimal 數據類型小數位數）0 當 DECIMAL/NUMBER 數據類型轉換成Avro類型數據時，指定的小數點后的位數。

配置步驟如下：

1、新建“QueryDatabaseTable”處理器

2、配置“SCHEDULING”調度時間

這里調度時間配置為99999s,讀取數據庫，這里讀取一次即可，默認0會不間斷讀取數據庫會對服務器造成非常大壓力。執行僅支持“Primary”主節點運行。

3、配置“PROPERTIES”

配置“Database Connection Pooling Service”選擇創建，在彈出頁面中可以按照默認選擇直接點擊“Create”。

點擊“->”繼續配置MySQL連接：

在彈出的頁面中填入：

連接MysqlURL:

jdbc:mysql://192.168.179.5:3306/mynifi?characterEncoding=UTF-8&useSSL=false

MySQL驅動類：com.mysql.jdbc.DriverMySQL jar包路徑：需要提前在NiFI集群各個節點上創建對應目錄并上傳jar包。連接mysql的用戶名和密碼。

通過以上配置好連接mysql如下：

配置其他屬性如下：

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

此處理器是將二進制Avro記錄轉換為JSON對象，提供了一個從Avro字段到JSON字段的直接映射，這樣得到的JSON將具有與Avro文檔相同的層次結構。輸出的JSON編碼為UTF-8編碼，如果傳入的FlowFile包含多個Avro記錄，則轉換后的FlowFile是一個含有所有Avro記錄的JSON數組或一個JSON對象序列(每個Json對象單獨成行)。如果傳入的FlowFile不包含任何記錄，則輸出一個空JSON對象。

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

配置項	默認值	允許值	描述
JSON container options（Json選擇）	array	nonearray	如何解析Json對象，none:解析Json將每個Json對象寫入新行。array：解析到的json存入JsonArray一個對象
Wrap Single Record（數據庫類型）	false	truefalse	指定解析到的空記錄或者單條記錄是否按照“JSON container options”配置包裝對象。
Avro schema（表名）			如果Avro數據沒有Schema信息，需要配置。

配置步驟如下：

1、創建“ConvertAvroToJSON”處理器

2、配置“PROPERTIES”

3、連接“QueryDatabaseTable”處理器和“CovertAvroToJSON”處理器

連接好兩個處理器后，可以配置“Connection”為負載均衡方式傳遞數據：

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

該處理器使用JsonPath表達式指定需要的Json數組元素，將Json數組中的多個Json對象切分出來，形成多個FlowFile。每個生成的FlowFile都由指定數組中的一個元素組成，并傳輸到關系"split"，原始文件傳輸到關系"original"。如果沒有找到指定的JsonPath，或者沒有對數組元素求值，則將原始文件路由到"failure"，不會生成任何文件。

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

配置項	默認值	允許值	描述
JsonPath Expression（Json表達式）			一個JsonPath表達式，它指定用以分割的數組元素。
Null Value Representation（Null值表示）	empty string	empty stringthe string "null"	指定結果為空值時的表示形式。

配置步驟如下：

1、創建“SplitJson”處理器

2、配置“PROPERTIES”

3、連接“ConvertAvroToJSON”處理器和“SplitJson”處理器

連接后，連接關系選擇“success”：

同時配置“ConverAvroToJSON”處理失敗的數據自動終止：

四、配置“PutHDFS”處理器

該處理器是將FlowFile數據寫入到HDFS分布式文件系統中。關于“PutHDFS”處理器的“Properties”主要配置的說明如下：

配置項	默認值	允許值	描述
Hadoop Configuration Resources（Hadoop配置）		nonearray	HDFS配置文件，一個文件或者由逗號分隔的多個文件。不配置將在ClassPath中尋找‘core-site.xml’或者‘hdfs-site.xml’文件。
Directory（目錄）			需要寫入文件的HDFS父目錄。如果目錄不存在，將創建該目錄。
Conflict Resolution Strategy(沖突解決)	fail	replaceignorefailappend	指示當輸出目錄中已經存在同名文件時如何處理。

配置步驟如下：

1、創建“PutHDFS”處理器

2、配置“PROPERTIES”

注意：以上需要在各個NiFi集群節點上創建“/root/test”目錄，并且在該目錄下上傳hdfs-site.xml和core-site.xml文件。

3、連接“SplitJson”處理器和“PutHDFS”處理器

同時設置“SplitJson”處理器中“failure”和“original”數據關系自動終止。

設置“PutHDFS”處理器“success”和“failure”數據關系自動終止：

配置好的連接關系如下：

五、??????????????運行測試

1、在MySQL創建庫“mynifi”,并且創建表“test1”，向表中插入10條數據

mysql> create database mynifi;Query OK, 1 row affected (0.02 sec)mysql> use mynifi;Database changedmysql> create table test1(id int,name varchar(255),age int );Query OK, 0 rows affected (0.07 sec)mysql> insert into test1 values (1,"zs",18),(2,"ls",19),(3,"ww",20),(4,"ml",21),(5,"tt",22)

復制

2、首先啟動“QueryDatabaseTable”處理器觀察隊列數據

3、單獨啟動“ConvertAvroToJson”處理器觀察隊列數據

4、單獨啟動“SplitJson”處理器觀察隊列數據

5、單獨啟動“PutHDFS”處理器觀察HDFS對應目錄數據

查看數據：

注意：

如果在“QueryDatabaseTable”處理器中設置增屬性“Maximum-value Columns”為id，那么每次查詢都是大于id的增量數據。如果想要存入HDFS文件為多行而不是一行，可以將“CovertAvroToJson”處理器屬性“JSON container options”設置為none，直接解析Avro文件得到一個個json數據，然后直接連接“PutHDFS”處理器即可。