1）案例需求：使用Flume監聽整個目錄的文件

2）需求分析：

3）實現步驟：

1．創建配置文件flume-dir-hdfs.conf

[root@henu1 job]# touch flume-dir-hdfs.conf [root@henu1 job]# vi flume-dir-hdfs.conf

添加以下內容：

a3.sources = r3 a3.sinks = k3 a3.channels = c3# Describe/configure the source a3.sources.r3.type = spooldir a3.sources.r3.spoolDir = /opt/flume/upload a3.sources.r3.fileSuffix = .COMPLETED a3.sources.r3.fileHeader = true #忽略所有以.tmp結尾的文件，不上傳 a3.sources.r3.ignorePattern = ([^ ]*\.tmp)# Describe the sink a3.sinks.k3.type = hdfs a3.sinks.k3.hdfs.path = hdfs://mycluster/flume/upload/%Y%m%d/%H #上傳文件的前綴 a3.sinks.k3.hdfs.filePrefix = upload- #是否按照時間滾動文件夾 a3.sinks.k3.hdfs.round = true #多少時間單位創建一個新的文件夾 a3.sinks.k3.hdfs.roundValue = 1 #重新定義時間單位 a3.sinks.k3.hdfs.roundUnit = hour #是否使用本地時間戳 a3.sinks.k3.hdfs.useLocalTimeStamp = true #積攢多少個Event才flush到HDFS一次 a3.sinks.k3.hdfs.batchSize = 100 #設置文件類型，可支持壓縮 a3.sinks.k3.hdfs.fileType = DataStream #多久生成一個新的文件 a3.sinks.k3.hdfs.rollInterval = 600 #設置每個文件的滾動大小大概是128M a3.sinks.k3.hdfs.rollSize = 134217700 #文件的滾動與Event數量無關 a3.sinks.k3.hdfs.rollCount = 0 #最小冗余數 a3.sinks.k3.hdfs.minBlockReplicas = 1# Use a channel which buffers events in memory a3.channels.c3.type = memory a3.channels.c3.capacity = 1000 a3.channels.c3.transactionCapacity = 100# Bind the source and sink to the channel a3.sources.r3.channels = c3 a3.sinks.k3.channel = c3

各參數解釋：

2. 啟動監控文件夾命令

/flume-ng agent --conf conf/ --name a3 --conf-file /opt/flume/job/flume-dir-hdfs.conf

說明：?在使用Spooling Directory Source時

不要在監控目錄中創建并持續修改文件

上傳完成的文件會以.COMPLETED結尾

被監控文件夾每500毫秒掃描一次文件變動

3. 向upload文件夾中添加文件

4. 查看HDFS上的數據

5. 等待1s，再次查詢upload文件夾

總結

以上是生活随笔為你收集整理的Flume实操（三）【实时读取目录文件到HDFS案例】的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。