1）案例需求：使用Flume監(jiān)聽整個(gè)目錄的文件

2）需求分析：

3）實(shí)現(xiàn)步驟：

1．創(chuàng)建配置文件flume-dir-hdfs.conf

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

添加以下內(nèi)容：

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結(jié)尾的文件，不上傳 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- #是否按照時(shí)間滾動(dòng)文件夾 a3.sinks.k3.hdfs.round = true #多少時(shí)間單位創(chuàng)建一個(gè)新的文件夾 a3.sinks.k3.hdfs.roundValue = 1 #重新定義時(shí)間單位 a3.sinks.k3.hdfs.roundUnit = hour #是否使用本地時(shí)間戳 a3.sinks.k3.hdfs.useLocalTimeStamp = true #積攢多少個(gè)Event才flush到HDFS一次 a3.sinks.k3.hdfs.batchSize = 100 #設(shè)置文件類型，可支持壓縮 a3.sinks.k3.hdfs.fileType = DataStream #多久生成一個(gè)新的文件 a3.sinks.k3.hdfs.rollInterval = 600 #設(shè)置每個(gè)文件的滾動(dòng)大小大概是128M a3.sinks.k3.hdfs.rollSize = 134217700 #文件的滾動(dòng)與Event數(shù)量無(wú)關(guān) a3.sinks.k3.hdfs.rollCount = 0 #最小冗余數(shù) 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

各參數(shù)解釋：

2. 啟動(dòng)監(jiān)控文件夾命令

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

說(shuō)明：?在使用Spooling Directory Source時(shí)

不要在監(jiān)控目錄中創(chuàng)建并持續(xù)修改文件

上傳完成的文件會(huì)以.COMPLETED結(jié)尾

被監(jiān)控文件夾每500毫秒掃描一次文件變動(dòng)

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

4. 查看HDFS上的數(shù)據(jù)

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

總結(jié)

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

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。