正文
一,前言
官方配置连接:点击进入
flume的使用是在任意的服务器进行agent启动。官方已经为我们写好了很多source,sink,channel的组件,仅需要我们按照官方提供的文档进行配置,然后启动即可。当然我们也可以自定制一下source和sink。在这我们就进行flume的演示,如有其他的需求,可以按照官方文档进行配置使用.
二,读取目录
下面是一个简单的读取目录示意图:
配置案例:
#定义三大组件的名称 ag1.sources = source1 ag1.sinks = sink1 ag1.channels = channel1 # 配置source组件 ag1.sources.source1.type = spooldir ag1.sources.source1.spoolDir = /root/log/ ag1.sources.source1.fileSuffix=.FINISHED ag1.sources.source1.deserializer.maxLineLength=5120 # 配置sink组件 ag1.sinks.sink1.type = hdfs ag1.sinks.sink1.hdfs.path =hdfs://hdp-01:9000/access_log/%y-%m-%d/%H-%M ag1.sinks.sink1.hdfs.filePrefix = app_log ag1.sinks.sink1.hdfs.fileSuffix = .log ag1.sinks.sink1.hdfs.batchSize= 100 ag1.sinks.sink1.hdfs.fileType = DataStream ag1.sinks.sink1.hdfs.writeFormat =Text ## roll:滚动切换:控制写文件的切换规则 ag1.sinks.sink1.hdfs.rollSize = 512000 ## 按文件体积(字节)来切 ag1.sinks.sink1.hdfs.rollCount = 1000000 ## 按event条数切 ag1.sinks.sink1.hdfs.rollInterval = 60 ## 按时间间隔切换文件 ## 控制生成目录的规则 ag1.sinks.sink1.hdfs.round = true ag1.sinks.sink1.hdfs.roundValue = 10 ag1.sinks.sink1.hdfs.roundUnit = minute ag1.sinks.sink1.hdfs.useLocalTimeStamp = true # channel组件配置 ag1.channels.channel1.type = memory ag1.channels.channel1.capacity = 500000 ## event条数 ag1.channels.channel1.transactionCapacity = 600 ##flume事务控制所需要的缓存容量600条event # 绑定source、channel和sink之间的连接 ag1.sources.source1.channels = channel1 ag1.sinks.sink1.channel = channel1
启动命令:
三,读取新增行
读取新增行示意图:
配置文件配置:
#定义三大组件的名称 ag1.sources = source1 ag1.sinks = sink1 ag1.channels = channel1 # 配置source组件 ag1.sources.source1.type = exec ag1.sources.source1.command = tail -F /root/log/app.log # 配置sink组件 ag1.sinks.sink1.type = hdfs ag1.sinks.sink1.hdfs.path =hdfs://hdp-01:9000/app_log/%y-%m-%d/%H-%M ag1.sinks.sink1.hdfs.filePrefix = app_log ag1.sinks.sink1.hdfs.fileSuffix = .log ag1.sinks.sink1.hdfs.batchSize= 100 ag1.sinks.sink1.hdfs.fileType = DataStream ag1.sinks.sink1.hdfs.writeFormat =Text ## roll:滚动切换:控制写文件的切换规则 ag1.sinks.sink1.hdfs.rollSize = 512000 ## 按文件体积(字节)来切 ag1.sinks.sink1.hdfs.rollCount = 1000000 ## 按event条数切 ag1.sinks.sink1.hdfs.rollInterval = 60 ## 按时间间隔切换文件 ## 控制生成目录的规则 ag1.sinks.sink1.hdfs.round = true ag1.sinks.sink1.hdfs.roundValue = 10 ag1.sinks.sink1.hdfs.roundUnit = minute ag1.sinks.sink1.hdfs.useLocalTimeStamp = true # channel组件配置 ag1.channels.channel1.type = memory ag1.channels.channel1.capacity = 500000 ## event条数 ag1.channels.channel1.transactionCapacity = 600 ##flume事务控制所需要的缓存容量600条event # 绑定source、channel和sink之间的连接 ag1.sources.source1.channels = channel1 ag1.sinks.sink1.channel = channel1
启动: