尚硅谷大数据技术NiFi教程-笔记02【NiFi（使用案例，同步文件、离线同步mysql数据到hdfs、实时监控kafka数据到hdfs）】_nifi使用除法

尚硅谷大数据技术-教程学习路线-笔记汇总表【课程资料下载】

• 视频地址：尚硅谷大数据NiFi教程（从部署到开发）_哔哩哔哩_bilibili

1. 尚硅谷大数据技术NiFi教程-笔记01【NiFi（基本概念、安装、使用）】
2. 尚硅谷大数据技术NiFi教程-笔记02【NiFi（使用案例，同步文件、离线同步mysql数据到hdfs、实时监控kafka数据到hdfs）】

目录

尚硅谷大数据NiFi教程（从部署到开发）

第3章-NiFi的使用

P10【10_案例一_添加处理器】14:08

P11【11_案例一_实现同步文件到hdfs】15:08

P12【12_案例二_添加mysql连接池服务】07:07

P13【13_案例二_离线读取mysql数据设置读取时间】08:06

P14【14_案例二优化_修改数据格式为json】04:30

P15【15_案例二优化_写入动态目录】10:18

P16【16_案例二优化_修改文件名控制文件的滚动】06:23

P17【17_案例三_生产消费kafka数据】10:36

P18【18_nifi总结】04:19

---

尚硅谷大数据NiFi教程（从部署到开发）

第3章-NiFi的使用

1. 同步文件，需求：同步本地磁盘文档上传到hdfs，通过NIFI自动监控磁盘文件上传到hdfs对应的文件夹。
2. 离线同步mysql数据到hdfs，需求：导出Mysql数据转换为Json串并保存到hdfs。
3. 实时监控kafka数据到hdfs，需求：实时监控kafka主题，将数据同步发送到hdfs。

P10【10_案例一_添加处理器】14:08

同步文件，需求：同步本地磁盘文档上传到hdfs，通过NIFI自动监控磁盘文件上传到hdfs对应的文件夹。

1. zk.sh start
2. /opt/module/nifi/nifi-1.19.1/bin/nifi.sh start
3. /opt/module/nifi/nifi-1.19.1/bin/nifi.sh status

This node is currently not connected to the cluster. Any modifications to the data flow made here will not replicate across the cluster.

解决方法：重启node1节点。

1. /opt/module/data   
2. [^\.].*   默认正则表达式，可以匹配所有文件。

/opt/module/nifi/data/dataInput

/opt/module/hadoop/hadoop-3.1.3/etc/hadoop/core-site.xml,/opt/module/hadoop/hadoop-3.1.3/etc/hadoop/hdfs-site.xml​​​​​

P11【11_案例一_实现同步文件到hdfs】15:08

[atguigu@node001 ~]$ cd /opt/module/applog/
[atguigu@node001 applog]$ java -jar gmall2020-mock-log-2021-10-10.jar
[atguigu@node001 applog]$ nohup java -jar gmall2020-mock-log-2021-10-10.jar &
 
[atguigu@node001 log]$ echo "你好！" >> 1.txt
[atguigu@node001 log]$ ll
总用量 764
-rw-rw-r-- 1 atguigu atguigu     10 5月  15 14:52 1.txt
-rw-rw-r-- 1 atguigu atguigu 776559 5月  15 14:52 app.2023-05-15.log
[atguigu@node001 log]$ 

想要时效性高，就将调度时间调低一些。

P12【12_案例二_添加mysql连接池服务】07:07

离线同步mysql数据到hdfs，需求：导出Mysql数据转换为Json串并保存到hdfs。

node001_mysql
jdbc:mysql://node001:3306/gmall
com.mysql.jdbc.Driver
 
/opt/jar
mysql-connector-java-5.1.27-bin.jar
/opt/jar/mysql-connector-java-5.1.27-bin.jar

P13【13_案例二_离线读取mysql数据设置读取时间】08:06

use gmall;
select * from activity_info
root
123456

更改ExecuteSQL的调度时间为86400秒，离线数仓的同步时间一般为一天同步一次，dataX全量同步。

P14【14_案例二优化_修改数据格式为json】04:30

P15【15_案例二优化_写入动态目录】10:18

需求：添加动态参数控制数据写入到hdfs的路径。

默认情况下所有的上传文件都会发送到同一个文件夹，导致文件混乱。

通过引入NiFi Expression Language函数可以动态修改最终导出的路径。具体内容可以参考官方网站Apache NiFi Expression Language Guide。

${now():format('yyyy-MM-dd')}
${now():format('yyyy-MM-dd HH:mm:ss')}
Apache NiFi Expression Language Guide

P16【16_案例二优化_修改文件名控制文件的滚动】06:23

${now():toNumber()}.data
${now():format('yyyy-MM-dd HH')}.data

多目录用来控制文件夹的滚动，文件名用来控制文件的滚动。

P17【17_案例三_生产消费kafka数据】10:36

实时监控kafka数据到hdfs，需求：实时监控kafka主题，将数据同步发送到hdfs。

node001:9092,node002:9092,node003:9092
nifi
/nifi_kafka/${dt}
${now():format('yyyy-MM-dd HH')}.kafka
 
kafka-console-producer.sh --broker-list node001:9092 --topic nifi
kafka-console-consumer.sh --bootstrap-server node001:9092 --topic nifiOutput

P18【18_nifi总结】04:19

实时监控mysql中变化的数据，Maxwell、flink-cdc。CDC是“Change Data Capture”的缩写。