大数据（一）flume

「爱情、让人受尽委屈。」 2023-03-12 15:28 155阅读 0赞

Flume是Cloudera提供的一个高可用的，高可靠的，分布式的海量日志采集、聚合和传输的系统，Flume支持在日志系统中定制各类数据发送方，用于收集数据；同时，Flume提供对数据进行简单处理，并写到各种数据接受方（可定制）的能力。

以下主要版本是 Flume-ng.

Flume的应用主要用于大数据方面，最开始开发出来也是用来对接KAFKA和HDFS。下图是大概是flume的结构图

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NocmVrMTE_size_16_color_FFFFFF_t_70][]

如上图：Flume大概概念有agent,source,sick,event等

**agent**： 为一个JVM进程，它以一个事件的形式将数据从源头送到目的，包括Source,Channel,Sink.

**source**:负责接收数据的组件，包括avro,exec,spooling directory,netct,syslog，http,kafka等等

常用到的有spooling directory,监听目录下的日志文件，收集日志，

avro， 监听本地端口的数据，多用于收集到汇集端

**channel**：source和sink之间的缓冲区，允许多个source和sink连接，channel是线程安全的，

channel有2种，一种是memory内存存储，性能高，但如果flume重启，数据会丢失，一个是file文件存储，需要I/O，

性能没有memory好，但数据不会丢失。

**sink**：  sink将channel的数据批量的写入到目的端。目的端包括hdfs,kafka,avro.file等。

sink是完全事务性的，在批量移除channel的数据前，会启动事务，一旦数据成功写入到目的端，事务将被提交，

这里确保数据不会被丢失。

event：数据传输的基本单元，以事件的形式将数据从源头传输到目的地。

flume安装： http://archive.apache.org/dist/flume，先去下载需要的版本

我这里下载的是1.7的版本

![20200514145616801.png][]

tar -zxvf apache-flume-1.7.0-bin.tar.gz -C /opt/module/

cd /opt/module/

mv apache-flume-1.7.0-bin flume

cd flume/conf

mv flume-env.sh.template flume-env.sh

vi flume-env.sh

\#指定 JAVAHOME

export JAVA\_HOME=/opt/module/jdk1.8.0\_144,安装完成。

之前就讲过，每启动Flume任务就是一个agent,agent是一个Jvm进程，里面包括source,channel,sink，

下面是一个Flume监听日志文件，然后发送kafka的配置文件

#define
    # a1 就是一个agent的别名，每个配置文件只能有1个
    a1.sources = r1   # 配置a1的source 别名 r1
    a1.sinks = k1     # 配置a1的sinks 别名 r1 多个输出用 空格 隔开
    a1.channels = c1  # 配置channels
    
    # source 数据来源配置 这里是监听本地某个日志文件
    a1.sources.r1.type = exec # r1的类型，可以为avro、thrift、exec、jms、spooling directory等
    a1.sources.r1.command = tail -F  /tmp/bigDate.log  # 上面为exec,这里就是监听一个日志文件，将刷新的日志收集
    a1.sources.r1.shell = /bin/bash -c  # 用bash执行
    
    # sink  数据输出落地端 这里配置的是输出到kafka
    a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
    a1.sinks.k1.kafka.bootstrap.servers = 10.0.100.129:6667  #kafkaip和端口
    a1.sinks.k1.kafka.topic = first    # 主题
    a1.sinks.k1.kafka.flumeBatchSize = 20  # 多少条推一次
    a1.sinks.k1.kafka.producer.acks = 1   
    a1.sinks.k1.kafka.producer.linger.ms = 1   # 延迟1毫秒
    
    # channel
    a1.channels.c1.type = memory   # channel的类型，有memory和file
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100  #事务
    
    # bind  关系互相绑定
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

启动指令：bin/flume-ng agent --conf conf/ --name a1 --conf-file job/group3/flume1-logger-flume.conf -Dflume.root.logger=INFO,console

这里的执行了配置文件flume1-logger-kafka.conf,   a1 是配置文件里agent的别名。

启动完之后，就可以往/tmp/bigDate.log里写数据，

ehco 'mytest' > /tmp/bigDate.log

每写一次，flume的source会收集变成event时间，发送到channel, sink 会监听channel（memory,数据放内存），

有数据过来，或者是20条，或者是1毫秒，将数据发送到kafka，这就是数据的整个流程.

当然在flume我们可以做一些简单的处理（拦截器），比如验证数据的格式，复制数据，将数据发送不同的sink端等

这里不建议去连接数据库或者是redis的地方，因为大数据，数据量大的话，做这些会让flume压力巨大

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NocmVrMTE_size_16_color_FFFFFF_t_70 1][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NocmVrMTE_size_16_color_FFFFFF_t_70]: /images/20230312/ed7aa05196034642a978e2004a57bb1f.png
[20200514145616801.png]: /images/20230312/f3010b33ea3e4fa3bd97ae2b22337b4b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NocmVrMTE_size_16_color_FFFFFF_t_70 1]: /images/20230312/5ecac278d87a481e8620677cb7412996.png