大数据NiFi（二十）：实时同步MySQL数据到Hive

此外，在“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中配置：
<!-- 配置hiveserver2 -->
<property>
  <name>hive.server2.thrift.port</name>
  <value>10000</value>
</property>
<property>
 <name>hive.server2.thrift.bind.host</name>
 <value>192.168.179.4</value>
</property>

​

2、在每台Hadoop 节点配置core-site.xml

<!-- 配置代理访问用户 -->
<property>     
 <name>hadoop.proxyuser.root.hosts</name>     
 <value>*</value>
 </property> 
<property>     
 <name>hadoop.proxyuser.root.groups</name>    
    <value>*</value> 
</property>

​

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]# beeline
beeline> !connect jdbc:hive2://node1:10000 root
Enter password for jdbc:hive2://node1:10000: 没有密码直接跳过即可
0: jdbc:hive2://node1:10000> show tables;
+------------------------------------+
|              tab_name              |
+------------------------------------+
| personinfo                         |
| test2                              |
+------------------------------------+