代码拉取完成,页面将自动刷新
接下来我们要做到如下的效果:
假设有一个目录 listendDir
,我们可以使用 flume 监听这个目录中的所有文件,例如新增一个文件 listendDir/test.txt
,那么 flume 就能自动将文件转移到 hadoop hdfs 中。
创建目录 mkdir ~/listendDir
在 job
目录下创建文件 listen-dir.conf
内容如下:
# listen-dir.conf
a1.sources = r1
a1.sources.r1.type = spooldir
# 这个是要监听的目录路径
a1.sources.r1.spoolDir = /home/master/listendDir
a1.sources.r1.fileHeader = true
a1.sources.r1.fileSuffix = .COMPLETED
a1.sources.r1.basenameHeader = true
a1.sources.r1.deletePolicy = immediate
a1.sources.r1.ignorePattern = ^\.
a1.sources.r1.deserializer = org.apache.flume.sink.solr.morphline.BlobDeserializer$Builder
a1.sinks = k1
a1.sinks.k1.type = hdfs
# 这个是 hdfs 中对应的目录路径
a1.sinks.k1.hdfs.path = hdfs://master:9000/listendDir
a1.sinks.k1.hdfs.filePrefix = events-
a1.sinks.k1.hdfs.fileSuffix = .log
a1.sinks.k1.hdfs.fileType = DataStream
a1.sinks.k1.hdfs.writeFormat = Text
a1.sinks.k1.hdfs.rollInterval = 0
a1.sinks.k1.hdfs.rollSize = 0
a1.sinks.k1.hdfs.rollCount = 10000
a1.sinks.k1.hdfs.batchSize = 100
a1.sinks.k1.hdfs.useLocalTimeStamp = true
a1.channels = c1
a1.channels.c1.type = memory
a1.channels.c1.capacity = 10000
a1.channels.c1.transactionCapacity = 1000
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
在 hdfs
上创建对应目录 /listendDir
hdfs dfs -mkdir /listendDir
bin/flume-ng agent -n a1 -c conf -f job/listen-dir.conf -Dflume.log.file -Dflume.root.logger=INFO,console
在 ~/listendDir
中添加 test.txt
文件,随便输入点内容那个,查看是否被同步到 hdfs
中
此处可能存在不合适展示的内容,页面不予展示。您可通过相关编辑功能自查并修改。
如您确认内容无涉及 不当用语 / 纯广告导流 / 暴力 / 低俗色情 / 侵权 / 盗版 / 虚假 / 无价值内容或违法国家有关法律法规的内容,可点击提交进行申诉,我们将尽快为您处理。