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环境)

1. 编写配置文件，参考官网例子

习惯上把任务的配置文件和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

2. 启动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

3. 启动结果显示端口已监听

[INFO - org.apache.flume.source.NetcatSource.start(NetcatSource.java:166)] Created serverSocket:sun.nio.ch.ServerSocketChannelImpl[/127.0.0.1:44444]

4. 通过netcat工具向端口发送数据

没有netcat telnet 也可以

nc localhost 44444 > Hello Flume OK

5. 查看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监控是拿到这个数据后用界面的方式展示出来了，相对比较直观。