Flume学习笔记
Flume [fluːm] (引水槽、运河) 王洪亮 2021/05/07 15:05 1. 简介 大数据技术解决的三个问题，海量数据的传输、存储和计算。Flume属于传输框架。专门传输日志的，多媒体不行。 2.安装 官网 http://flume.apache.org/ 下载apache-flume-1.9.0-bin.tar.gz，下载后解压即可，类似tomcat中间件，需要改配置文件 ， 把apache-flume-1.9.0-bin/conf/flume-env.sh.template改成flume-env.sh，里面配置jdk地址，安装完毕。 export JAVA_HOME=D:\sde\Java\jdk1.8.0_131 3.Hello Flume样例 flume是按任务启动的，这个任务在flume这里叫agent，每个agent分为三块，source(对接数据来源)、channel(中间缓冲区)、sink(数据去向)。任务过程简述就是从一个地方采集数据后送到其他地方。 此次案例source数据采集自端口，通过channel再经过sink将数据输出到终端屏幕上。 开始编写Hello Flume样例(win环境) 编写配置文件，参考官网例子 习惯上把任务的配置文件和conf区分开，这里新建jobs目录下新建netcat-flume-logger.conf,命名格式最好区分出source和sink的类型 #完整的conf分几块，变量声明区、source区、interceptor区、channelSelector区、channel区、sink区、关系绑定区 #变量声明区（必须） a1指的是agent的变量名，启动时指定 #分别声明 sources、channels、sinks、sinkgroups等 a1.sources = r1 a1.channels = c1 a1.sinks = k1 #source区（必须） 指定source类型 具体类型需要看官网文档source 部分 a1.sources.r1.type = netcat a1.sources.r1.bind = localhost a1.sources.r1.port = 44444 #interceptor区(可选),代码自写,配合multiplexing channelSelector使用 #channelSelector区（可选），不写默认是replicating，把数据发给每个sink a1.sources.r1.selector.type=replicating #channel区（必须） 指定channel类型 具体类型需要看官网文档channel部分 a1.channels.c1.type = memory #sink处理器区，声明sink组，选择处理器是故障转移还是负载均衡，不写就是默认 #sink区（必须） 指定sink类型 具体类型需要看官网文档sink部分 a1.sinks.k1.type = logger #关系绑定区（必须） a1.sources.r1.channels = c1 a1.sinks.k1.channel = c1 启动flume的agent #win环境启动命令 进入到bin目录 cmd flume-ng agent --conf ../conf --conf-file ../jobs/netcat-flume-logger.conf --name a1 -property flume.root.logger=INFO,console #简写 flume-ng agent -c ../conf -f ../jobs/netcat-flume-logger.conf -n a1 -property flume.root.logger=INFO,console #linux环境启动命令 那个logger不一样 #-Dflume.root.logger=INFO,console 启动结果显示端口已监听 [INFO - org.apache.flume.source.NetcatSource.start(NetcatSource.java:166)] Created serverSocket:sun.nio.ch.ServerSocketChannelImpl[/127.0.0.1:44444] 通过netcat工具向端口发送数据 没有netcat telnet 也可以 nc localhost 44444 > Hello Flume OK 查看flume的日志输出 [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:95)] Event: { headers:{} body: 48 65 6C 6C 6F 20 46 6C 75 6D 65 Hello Flume } 4.Source种类 拉出几个常用的。其他的查看官方文档 1.Avro Source Flume集群使用，多个Fulme连接需要使用Avro的source和sink连接，自身相当于服务端，需要优先启动 #相关source区 写法 a1.sources.r1.type=avro a1.sources.r1.bind=0.0.0.0 a1.sources.r1.port=4141 2. Exec Source 用命令监控某个文件的变化，将变化的内容传送走，适用于监控一个实时变化的日志文件，缺点是不能断点续传，容易数据丢失或是数据重复 #相关source区 写法 a1.sources.r1.type=exec #后跟 监控文件的命令 a1.sources.r1.command=tail -F /var/log/secure 3. Spooling Directory Source 监控本地文件夹，文件夹中新增文件时，会将整个文件内容取走，并加后缀做标记（.COMPLETED），表示已经取过了，标记后的文件不在处理，内容变化也不管。适用于批量上传，缺点是不能适应动态变化的文件 #相关source区 写法 a1.sources.r1.type=spooldir #后跟监控文件的目录 a1.sources.r1.spoolDir=/var/log/apache/flumeSpool 4. Taildir Source 可以监控不同文件夹下的不同文件。支持断点续传 #相关source区 写法 a1.sources.r1.type = TAILDIR #指定保存断点续传的文件 a1.sources.r1.positionFile = /var/log/flume/position.json #指定文件组 可以多个 a1.sources.r1.filegroups = f1 f2 #指定 单个文件 a1.sources.r1.filegroups.f1 = /var/log/test1/example.log #指定多个文件 * 不能出现在第一位 a1.sources.r1.filegroups.f2 = /var/log/test2/.*log.* 5. NetCat TCP Source 监听TCP端口用的，还有个UDP Source 就type不一样 #相关source区 写法 udp 的type是netcatudp a1.sources.r1.type=netcat a1.sources.r1.bind=0.0.0.0 a1.sources.r1.port=6666 5.Channel种类 常用的Memory Channel和File Channel 1. Memory Channel 将数据缓存在内存中，快，闪断丢数据 #channel区 a1.channels.c1.type=memory 2. File Channel 将数据缓存在本地文件中，慢点，稳定 #channel区 a1.channels.c1.type=file #数据存放的目录，可以有多个目录，逗号分割 a1.channels.c1.dataDirs=/mnt/flume/data 6.Sink种类 常用的拉出 1. Avro Sink Flume集群收尾连接，相对source 而言这个算客户端，去绑定服务端IP和端口 #sink 区 a1.sinks.k1.type = avro a1.sinks.k1.hostname = 10.10.10.10 a1.sinks.k1.port = 4545 2. Logger Sink 日志打印在终端屏幕上，需要agent启动的时候搭配对应命令 问题：中文没打出来；maxBytesToLog这玩意不好使 #sink区 #-property flume.root.logger=INFO,console #-Dflume.root.logger=INFO,console a1.sinks.k1.type=logger 3. File Roll Sink 日志输出到本地文件，问题是就算没有接收到数据，也会不停的生成新的空文件 #sink区 a1.sinks.k2.type = file_roll #指定生成文件夹路径，文件名会自动生成 a1.sinks.k2.sink.directory = ../data/ #指定新文件生成频率，30s a1.sinks.k2.sink.rollInterval = 30 7.Channel Selector 这个就俩，算自定义的三个 1. Replicating Channel Selector (default) 默认的，可以不写，把数据发给所有sink #channel selector 区 a1.sources.r1.selector.type=replicating 2. Multiplexing Channel Selector 根据拦截器规则，选择发送到哪个sink #channel selector 区 a1.sources.r1.selector.type = multiplexing #这个关键变量，拦截器里写的，写什么是什么，不一定是state a1.sources.r1.selector.header = state #根据这个state的值选择sink,如果header中的state=CZ就发个c1 a1.sources.r1.selector.mapping.CZ = c1 # 等于US就给c2 c3 a1.sources.r1.selector.mapping.US = c2 c3 8.Sink Processor 目前算默认的三种，默认的数据给谁，谁就发走 1. Failover Sink Processor 故障转移，根据优先级，可着最高优先级的用，挂了再用优先级低的 #sink处理器区 #这个算声明，可以拿到声明区 a1.sinkgroups = g1 a1.sinkgroups.g1.sinks = k1 k2 #选择处理器类型为 故障转移 a1.sinkgroups.g1.processor.type = failover #指定sink组成员优先级 a1.sinkgroups.g1.processor.priority.k1 = 5 #最高等级k2一直使用，直到挂了为止，才能轮到k1 a1.sinkgroups.g1.processor.priority.k2 = 10 2. Load balancing Sink Processor 负载均衡，可选随机或是均衡发送 #sink处理器区 #这个算声明，可以拿到声明区 a1.sinkgroups = g1 a1.sinkgroups.g1.sinks = k1 k2 #选择处理器类型为 负载均衡 a1.sinkgroups.g1.processor.type = load_balance #指定模式为随机发送，还可以选round_robin均衡发送 a1.sinkgroups.g1.processor.selector = random 9.自定义Inteceptor idea新建maven项目开整 \> hello 1. 引入依赖 <dependency> <groupId>org.apache.flume</groupId> <artifactId>flume-ng-core</artifactId> <version>1.9.0</version> </dependency> 2. 写自己的拦截器 package com.flume; import org.apache.flume.Context; import org.apache.flume.Event; import org.apache.flume.interceptor.Interceptor; import java.util.HashMap; import java.util.List; import java.util.Map; public class MyInterceptor implements Interceptor { @Override public void initialize() { System.out.println("---------initialize---------"); } @Override public Event intercept(Event event) { String body = new String(event.getBody()); Map<String, String> map = new HashMap<>(); if(body.contains("hello")){ map.put("myStatus","up"); event.setHeaders(map); }else{ map.put("myStatus","down"); event.setHeaders(map); } return event; } @Override public List<Event> intercept(List<Event> list) { return null; } @Override public void close() {} public static class MyBuilder implements Interceptor.Builder{ @Override public Interceptor build() { return new MyInterceptor(); } @Override public void configure(Context context) {} } } 3. 项目打个jar包放到Flume的lib目录下 4. 编写Flume配置文件 netcat-flume-doubleLogger.conf，本来想两个sink都发到logger的，后来发现终端上分不出哪个是哪个，把其中一个sink改到落到本地文件了 #变量区 a1.sources = r1 a1.channels = c1 c2 a1.sinks = k1 k2 #source区 a1.sources.r1.type = netcat a1.sources.r1.bind = localhost a1.sources.r1.port = 44444 #intercetor区 a1.sources.r1.interceptors = i1 a1.sources.r1.interceptors.i1.type = com.flume.MyInterceptor$MyBuilder #channel selector 区 a1.sources.r1.selector.type = multiplexing #我代码里定义的是 myStatus a1.sources.r1.selector.header = myStatus #myStatus=up就发个c1 a1.sources.r1.selector.mapping.up = c1 # 等于down就给c2 a1.sources.r1.selector.mapping.down = c2 #channel区 a1.channels.c1.type = memory a1.channels.c2.type = memory #sink区 a1.sinks.k1.type = logger a1.sinks.k2.type = file_roll a1.sinks.k2.sink.directory = ../data/ #绑定区 a1.sources.r1.channels = c1 c2 a1.sinks.k1.channel = c1 a1.sinks.k2.channel = c2 5. 启动 flume-ng agent --conf ../conf --conf-file ../jobs/netcat-flume-doubleLogger.conf --name a1 -property flume.root.logger=INFO,console 6. 测试效果 启动netcat发送命令 > nc localhost 44444 6351 OK hello OK 一个在终端屏幕上有显示，一个在本地文件中有显示。 10自定义Source 1. 依赖 <dependency> <groupId>org.apache.flume</groupId> <artifactId>flume-ng-core</artifactId> <version>1.9.0</version> </dependency> 2. 代码 package com.flume; import org.apache.flume.Context; import org.apache.flume.Event; import org.apache.flume.PollableSource; import org.apache.flume.conf.Configurable; import org.apache.flume.event.SimpleEvent; import org.apache.flume.source.AbstractSource; public class MySource extends AbstractSource implements Configurable, PollableSource { private String aaa; private String bbb; /** * 定义配置文件中的变量 * @param context */ @Override public void configure(Context context) { aaa = context.getString("aaa"); //没取到值 就给默认值 bbb = context.getString("bbb", "GangZi"); } @Override public Status process(){ Status status = null; try { //创建个事件 Event e = new SimpleEvent(); String content = aaa+"->"+bbb+": "+Math.random(); e.setBody(content.getBytes("utf-8")); //把event传给channel getChannelProcessor().processEvent(e); status = Status.READY; } catch (Throwable t) { status = Status.BACKOFF; if (t instanceof Error) { throw (Error)t; } } try { //睡两秒 要不刷的太快了 Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); } return status; } @Override public long getBackOffSleepIncrement() { return 0; } @Override public long getMaxBackOffSleepInterval() { return 0; } } 3. 写配置文件custom-flume-logger.conf #变量区 a1.sources = r1 a1.channels = c1 a1.sinks = k1 #source区 a1.sources.r1.type = com.flume.MySource a1.sources.r1.aaa = LiLei #a1.sources.r1.bbb = HanMeiMei #channel区 a1.channels.c1.type = memory #sink区 a1.sinks.k1.type = logger a1.sinks.k1.maxBytesToLog = 18 #绑定区 a1.sources.r1.channels = c1 a1.sinks.k1.channel = c1 4. 启动命令 flume-ng agent --conf ../conf --conf-file ../jobs/custom-flume-logger.conf --name a1 -property flume.root.logger=INFO,console 5. 查看效果 [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:95)] Event: { headers:{} body: 4C 69 4C 65 69 2D 3E 47 61 6E 67 5A 69 3A 20 30 LiLei->GangZi: 0 } 11自定义Sink 1. 依赖 <dependency> <groupId>org.apache.flume</groupId> <artifactId>flume-ng-core</artifactId> <version>1.9.0</version> </dependency> 2. 代码 package com.flume; import org.apache.flume.*; import org.apache.flume.conf.Configurable; import org.apache.flume.sink.AbstractSink; public class MySink extends AbstractSink implements Configurable { private String ccc; private String ddd; @Override public void configure(Context context) { ccc = context.getString("ccc"); ddd = context.getString("ddd","xxxxxxxx"); } @Override public Status process() throws EventDeliveryException { Status status = null; Channel ch = getChannel(); Transaction txn = ch.getTransaction(); txn.begin(); try { Event event = ch.take(); if(event == null){ txn.rollback(); status = Status.BACKOFF; }else { byte[] body = event.getBody(); String content = new String(body); System.out.println("ccc="+ccc); System.out.println("content="+content); System.out.println("ddd="+ddd); txn.commit(); status = Status.READY; } } catch (Exception e) { e.printStackTrace(); txn.rollback(); status = Status.BACKOFF; }finally { txn.close(); } return status; } } 3. flume配置文件netcat-flume-custom.conf #变量区 a1.sources = r1 a1.channels = c1 a1.sinks = k1 #source区 a1.sources.r1.type = netcat a1.sources.r1.bind = localhost a1.sources.r1.port = 44444 #channel区 a1.channels.c1.type = memory #sink区 a1.sinks.k1.type = com.flume.MySink a1.sinks.k1.ccc =--hello-- a1.sinks.k1.ddd =--moto-- #绑定区 a1.sources.r1.channels = c1 a1.sinks.k1.channel = c1 4. 启动agent flume-ng agent --conf ../conf --conf-file ../jobs/netcat-flume-custom.conf --name a1 -property flume.root.logger=INFO,console 5. 客户端发送命令 telnet localhost 44444 > 1 OK 6. flume终端打印 ccc=--hello-- content=1 ddd=--moto-- 12.Flume监控工具 下面的实验，windows环境下怎么跟多个参数 不知道 1.Http监控，json形式文本 使用这种监控方式，只需要在启动flume的时候在启动参数上面加上监控配置，例如这样： bin/flume-ng agent --conf conf --conf-file conf/flume_conf.properties --name collect -Dflume.monitoring.type=http -Dflume.monitoring.port=1234 其中-Dflume.monitoring.type=http表示使用http方式来监控，后面的-Dflume.monitoring.port=1234表示我们需要启动的监控服务的端口号为1234，这个端口号可以自己随意配置。然后启动flume之后，通过http://ip:1234/metrics就可以得到flume的一个json格式的监控数据。 2.ganglia监控，图形化界面 这种监控方式需要先安装ganglia然后启动ganglia，然后再启动flume的时候加上监控配置，例如： bin/flume-ng agent --conf conf --conf-file conf/producer.properties --name collect -Dflume.monitoring.type=ganglia -Dflume.monitoring.hosts=ip:port 其中-Dflume.monitoring.type=ganglia表示使用ganglia的方式来监控，而-Dflume.monitoring.hosts=ip:port表示ganglia安装的ip和启动的端口号。 flume监控还可以使用zabbix，但是这种方式需要在flume源码中添加监控模块，相对比较麻烦，由于不是flume自带的监控方式，这里不讨论这种方式。 因此，flume自带的监控方式其实就是http、ganglia两种，http监控只能通过一个http地址访问得到一个json格式的监控数据，而ganglia监控是拿到这个数据后用界面的方式展示出来了，相对比较直观。
