flume描述

flume用途

flume基本组件

配置flume

搭建环境：

解压flume安装包将其放入到opt/soft目录

在/opt/soft目录下将apache-flume-1.9.0-bin.tar.gz 改名为flume190

到/opt/soft/flume190/conf目录中将临时配置文件flume-env.sh.template拷贝为配置文件flume-env.sh

继续在conf目录下配置flume-env.sh文件（修改22行和25行）

在conf目录下新建netcat-logger.conf文件

编辑安装nc（在任何目录安装都行）

启动nc

安装拨号（在任何目录安装都行）

拨号测试

flume拨号测试

flumedemo.log中追加数据

tail显示内容变动

将指定文件中的内容输出到控制台（linux系统内部的文件）

编辑conf文件

编辑启动flume

插入数据进入flumedemo.log文件

将指定文件中的内容输出到控制台（window系统文件传入到linux系统的文件）

导入数据csv文件

编辑conf文件

修改文件名来符合在上面定义的规则

启动flume

flume描述

flume用途

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

flume基本组件

(1) Flume 工作流程
Source 采集数据并包装成Event，并将Event缓存再Channel中，Sink不断地从Channel 获取Event，并解决成数据，最终将数据写入存储或索引系统

(2) Agent
Agent 是一个JVM进程，它以事件的形式将数据从源头送至目的。是 Flume 数据传输的基本单元

Agent 的3个组件的设计思想，主要考虑的是：source 和 sink 之间解耦合，以及异步操作。

每一个agent相当于一个数据（被封装成Event对象）传递员，内部有3个核心组件：

Source：采集组件，用户跟数据源对接，以获取数据；它有各种各样的内置实现；
Sink：下沉组件，用于往下一级agent 传递数据或者向最终存储系统传递数据。
Channel : 传输通道组件，用于从source将数据传递到sink
(3) Source
Source 是负责接收数据到Flume Agent的组件，采集数据并包装成Event。Source组件可以处理各种类型、各种格式的日志数据，包括avro、thrift、exec、jms、spooling directory、netcat、sequence generator、syslog、http、legacy

(4) Sink
Sink 不断地轮询Channel 中的事件且批量地移除他们，并将这些事件批量写入到存储或索引系统、或者被发送到另一个Flume Agent。

**Sink是完全事务性的。**在从 Channel 批量删除数据之前，每个 Sink 用 Channel 启动一个事务。批量时间一旦成功写出到存储系统或下一个 Flume Agent，Sink 就利用Channel 提交事务。事务一旦被提交，该 Channel 从自己的内部缓存区删除事件。

Sink 组件目的地包括hdfs、logger、avro、thrift、ipc、file、HBase、solr、自定义

(5) Channel
Channnel 是位于 Source 和 Sink 之间的缓冲区。因此，Channel 允许Source 和 Sink 运作在不同的速率上。Channel是线程安全的，可以同时处理几个Source的写入操作和几个Sink的读取操作

Flume 自带两种Channel：

Memory Channel是内存中的队列。Memory Channel在不需要关系数据丢失的情景下适用。如果需要关心数据丢失，那么Memory Channel就不应该使用，因为程序死亡、机器当即或者重启都会导致数据丢失
File Channel 将所有事件写到磁盘。因此在程序关闭或机器宕机的情况下不会丢失数据
(6) Event
数据在channel中的封装形式。

传输单元，Flume数据传输的基本单元，以Event的形式将数据从源头送至目的地。Event 由 Header 和Body 两部分组成，Header用来存放该event的一些属性，为K-V结构，Body 用来存放该条数据，形式为字节数组

(7) interceptor 拦截器
拦截器工作在source 组件之后，source 产生的 event 会被出啊还能入拦截器根据需要进行拦截处理，而且，拦截器可以组成拦截器链！

拦截器在flume中有一些内置的功能；

用户也可以根据自己的数据处理需求，自己开发自定义拦截器，这也是Flume的一个可以用来自定义扩展的接口。

(8) channel selector
一个source可以对接多个channel ，则 event 在这 n 个channel 之间传递的策略，由配置的channel selector 决定；

channel selector 有两种实现：replicating（复制），multiplexing（多路复用）

(9) sink processor
如果sink 和 channel 是一对一关系，则不需要专门的sink processor；

如果要配置一个channel 对多个sink ，则需要将这多个sink配置成一个sink group （sink组）；

event 在一个组中的多个sink 间如何传递，则由所配置的sink processor l来决定；

原文链接：https://blog.csdn.net/weixin_45866849/article/details/125729411

配置flume

搭建环境：

基于flume190 +hadoop313+hbase235 +hive312 +jdk180 +sqoop147 +zeppelin +zk235

解压flume安装包将其放入到opt/soft目录

[root@gree2 install]# tar -zxf ./apache-flume-1.9.0-bin.tar.gz -C ../soft/

在/opt/soft目录下将apache-flume-1.9.0-bin.tar.gz 改名为flume190

[root@gree2 soft]# mv apache-flume-1.9.0-bin/ flume190

到/opt/soft/flume190/conf目录中将临时配置文件flume-env.sh.template拷贝为配置文件flume-env.sh

[root@gree2 conf]# cp flume-env.sh.template flume-env.sh

继续在conf目录下配置flume-env.sh文件（修改22行和25行）

[root@gree2 conf]# vim flume-env.sh

export JAVA_HOME=/opt/soft/jdk180
export JAVA_OPTS="-Xms2000m -Xmx2000m -Dcom.sun.management.jmxremote"

在conf目录下新建netcat-logger.conf文件

[root@gree2 conf]# vim ./netcat-logger.conf

填入下列信息

a1.sources=r1
a1.sinks=k1
a1.channels=c1

a1.sources.r1.type=netcat
a1.sources.r1.bind=192.168.136.20
a1.sources.r1.port=8888

a1.channels.c1.type=memory
a1.channels.c1.capacity=1000
a1.channels.c1.transactionCapacity=100

a1.sinks.k1.type=logger

a1.sources.r1.channels=c1
a1.sinks.k1.channel=c1

安装nc（在任何目录安装都行）

[root@gree2 conf]# yum install -y nc

启动nc

[root@gree2 conf]# nc -lk 8888

安装拨号协议（在任何目录安装都行）

[root@gree2 ~]# yum list telnet*

[root@gree2 ~]# yum install telnet-server

[root@gree2 ~]# yum install telnet.*

拨号测试

[root@gree2 ~]# telnet localhost 8888

flume拨号测试

开启拨号服务

[root@gree2 flume190]# ./bin/flume-ng agent --name a1 --conf ./conf/ --conf-file ./conf/netcat-logger.conf -Dflume.root.logger=INFO,console

[root@gree2 ~]# telnet 192.168.61.146 8888

flumedemo.log中追加数据

[root@gree2 ~]# echo 'hello'>./flumedemo.log

[root@gree2 ~]# echo 'world' >> ./flumedemo.log

[root@gree2 ~]# echo 'happy birthday' >> ./flumedemo.log

tail显示内容变动

[root@gree2 flume190]# tail -f /root/flumedemo.log

将指定文件中的内容输出到控制台（linux系统内部的文件）

编辑conf文件

[root@gree2 flume190]# vim ./conf/file-flume-logger.conf

#将指定文件中的内容输出到控制台
a2.sources=r1
a2.sinks=k1
a2.channels=c1

a2.sources.r1.type=exec
a2.sources.r1.command=tail -f /root/flumedemo.log

a2.channels.c1.type=memory
a2.channels.c1.capacity=1000
a2.channels.c1.transactionCapacity=100

a2.sinks.k1.type=logger

a2.sources.r1.channels=c1
a2.sinks.k1.channel=c1

启动flume

[root@gree2 flume190]# ./bin/flume-ng agent --name a2 --conf ./conf/ --conf-file ./conf/file-flume-logger.conf -Dflume.root.logger=INFO,console

插入数据进入flumedemo.log文件

[root@gree2 ~]# echo 'sh very nice' >> ./flumedemo.log

将指定文件中的内容输出到控制台（window系统文件传入到linux系统的文件）

导入数据csv文件

编辑conf文件

[root@gree2 flume190]# vim ./conf/events-flume-logger.conf

填入下列代码

events.sources=eventsSource
events.channels=eventsChannel
events.sinks=eventsSink

events.sources.eventsSource.type=spooldir
events.sources.eventsSource.spoolDir=/opt/flumelogfile/events
events.sources.eventsSource.deserializer=LINE
events.sources.eventsSource.deserializer.maxLineLength=32000
events.sources.eventsSource.includePattern=events_[0-9]{4}-[0-9]{2}-[0-9]{2}.csv

events.channels.eventsChannel.type=file
events.channels.eventsChannel.checkpointDir=/opt/flumelogfile/checkpoint/events
events.channels.eventsChannel.dataDir=/opt/flumelogfile/data/events

events.sinks.eventsSink.type=logger

events.sources.eventsSource.channels=eventsChannel
events.sinks.eventsSink.channel=eventsChannel