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