flume-ng的简单使用

小鱼儿 2022-07-14 01:28 164阅读 0赞

首先说下怎么安装吧，flume-ng的安装，flume-ng安装特别简单  
1：首先下载安装包 解压，然后进入到conf目录下，首先把flume-env.sh.template文件修改名字为flume-env.sh，然后修改flume-env.sh里面的JAVA\_HOME的地址，

2：测试 bin/flume-ng version  
如果能出现版本信息就说明安装成功了，接下来就可以使用了

使用的时候主要是一个配置文件 下面是一个把一个文件夹下面的文件都上传到hdfs上  
在conf文件下面创建了一个文件 localToHdfs.conf文件  
在localToHdfs.conf文件里面配置以下内容

#定义agent名， source、channel、sink的名称
    a1.sources = r1
    a1.channels = c1
    a1.sinks = k1
    
    #具体定义source
    a1.sources.r1.type = spooldir
    a1.sources.r1.spoolDir = /usr/logs
    
    #具体定义channel
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 10000
    a1.channels.c1.transactionCapacity = 100
    
    #定义拦截器，为消息添加时间戳
    a1.sources.r1.interceptors = i1
    a1.sources.r1.interceptors.i1.type = org.apache.flume.interceptor.TimestampInterceptor$Builder
    
    #具体定义sink
    a1.sinks.k1.type = hdfs
    a1.sinks.k1.hdfs.path = hdfs://master:9000/flume/%Y%m%d
    a1.sinks.k1.hdfs.filePrefix = events-
    a1.sinks.k1.hdfs.fileType = DataStream
    
    #不按照条数生成文件
    a1.sinks.k1.hdfs.rollCount = 0
    #HDFS上的文件达到128M时生成一个文件
    a1.sinks.k1.hdfs.rollSize = 1342177
    #HDFS上的文件达到60秒生成一个文件
    a1.sinks.k1.hdfs.rollInterval = 1200
    
    #组装source、channel、sink
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

然后启动，命令为

bin/flume-ng agent -n a1 -c conf -f conf/test.conf -Dflume.root.logger=INFO,console

解释  
\-n 指定agent名称

\-c 指定配置文件目录

\-f 指定配置文件

\-Dflume.root.logger=DEBUG,console

好了，已经完成了，赶快去测试测试吧

其他的配置就直接去官网搜索吧

再分享几个吧  
这个是监控一个文件的内容增加的，把监控到的内容打印到控制台上

#bin/flume-ng agent -n a2 -f /home/hadoop/a2.conf -c conf -Dflume.root.logger=INFO,console
    #定义agent名， source、channel、sink的名称
    a2.sources = r1
    a2.channels = c1
    a2.sinks = k1
    
    #具体定义source
    a2.sources.r1.type = exec
    a2.sources.r1.command = tail -F /home/hadoop/a.log
    
    #具体定义channel
    a2.channels.c1.type = memory
    a2.channels.c1.capacity = 1000
    a2.channels.c1.transactionCapacity = 100
    
    #具体定义sink
    a2.sinks.k1.type = logger
    
    #组装source、channel、sink
    a2.sources.r1.channels = c1
    a2.sinks.k1.channel = c1

监听网络端口

#定义agent名， source、channel、sink的名称
    a1.sources = r1
    a1.channels = c1
    a1.sinks = k1
    
    #具体定义source
    a1.sources.r1.type = netcat
    a1.sources.r1.bind = localhost
    a1.sources.r1.port = 8888
    
    #具体定义channel
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    
    #具体定义sink
    a1.sinks.k1.type = logger
    
    #组装source、channel、sink
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

遗留的问题：  
同一个source（源头）怎么才能发送到多个接收方  
还有就是怎么才能把多个source发动给同一个sink

真正上线的时候容易oom（out of memory） 怎么解决?配置初始化的堆内存信息