Flume是一种分布式的、高可靠的和高可用的服务，用于有效地收集、聚合和移动大量日志数据框架。
（1）支持自定义Source
flume 支持在日志系统中定制各类数据发送方，用于收集数据。
（2）支出数据简单处理
flume支持对数据进行简单处理，并写到各种数据接受方(比如文本、HDFS、Hbase等)的能力 。
（3）事件基本数据单位
flume 的数据流由事件(Event)贯穿始终。事件是 Flume 的基本数据单位，它携带日志数据(字节数组形式)并且携带有头信息，这些 Event 由 Agent 外部的 Source 生成，当 Source 捕获事件后会进行特定的格式化，然后 Source 会把事件推入(单个或多个) Channel 中。可以把 Channel 看作是一个缓冲区，它将保存事件直到 Sink 处理完该事件。
（4）高可靠性
当节点出现故障时，日志能够被传送到其他节点上而不会丢失。Flume提供了三种级别的可靠性保障，从强到弱依次分别为：
实时监控读取服务器产生到本地磁盘的数据，将数据写入到HDFS、Kafka等下游处理流程中去。
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-yFHTtXjw-1652500698450)
Event是Flume的数据传输最基本单位，flume在数据传输过程中是使用Event将数据封装起来进行传输的。
本质：生产的数据，可以是日志记录、 avro 对象等，如果是文本文件通常是一行记录。
组成：Event由Header和Body，Header使用k-v存放的Event的属性信息；Body以字节数组形成存储该条信息。
Event: { headers:{} body: 68 61 64 6F 6F 70 hadoop }
Agent是数据传输（接受-缓存-发送）形式，Agent是一个JVM进程，flume 以 Agent 为最小的独立运行单位，包含最基本的三个组件：Source、Chanel、Sink。
Source是负责接收数据到Flume Agent的组件，Source组件可以处理各种类型、各种格式的日志数据，包括avro、thrift、exec、spooling directory、netcat、tailder…以及Custom Source。
Channel是为了Source和Sink之间的缓冲区。Channel允许Source和Sink运作在不同的速率上。Channel是线程安全的，可以同时处理多个Source的写入操作、多个Sink的读取操作。
Sink不断轮询Channel中的事件且批量移除它们，并将这些事件批量写入到存储或者索引系统、或者被发送到到另一个Flume Agent。
（1）每个web server产生日志文件
（2）每一台web server所对应的节点上开启一个进程，分别对应Agent1，Agent2、Agent3；
Avro Sources将数据序列化，写入到channel，之后将数据送往Avro Sink，以Avro sink的形式将数据送往下游进行处理。
（3）下游的avro Source接受上游的Avro Sink的数据，在经过下游的Flume汇总到HDFS中。
https://www.apache.org/dyn/closer.lua/flume/1.9.0/apache-flume-1.9.0-bin.tar.gz
tar -xzvf apache-flume-1.9.0-bin.tar.gz -C /opt/module/flume
sudo vim /etc/profile.d/my_env.sh
export FLUME_HOME=/opt/module/flume/apache-flume-1.9.0-bin
export PATH=$PATH:$FLUME_HOME/bin
source /etc/profile.d/my_env.sh
flume-ng version
rm /lib/guava-11.0.2.jar
使用 Flume 监听一个端口，收集该端口数据，并打印到控制台。
（1）通过netcat工具向本机的4444端口发送数据；
（2）Flume监控本机的4444端口通过Flume的Source端读取数据；
（3）Flume将获取数据通过Sink端写到控制台。
（1）安装 netcat 工具
sudo yum install -y nc
（2）判断 44444 端口是否被占用
sudo netstat -nlp | grep 44444
（3）在 flume 目录下创建 job 文件夹并进入 job 文件夹。
mkdir job
cd job/
（4）在 job 文件夹下创建 Flume Agent 配置文件 flume-netcat-logger.conf。
vim flume-netcat-logger.conf
（6）在 flume-netcat-logger.conf 文件中添加如下内容。
添加内容如下：
# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44444
# Describe the sink
a1.sinks.k1.type = logger
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
（6）开启Flume监听端口
bin/flume-ng agent --conf conf/ --name a1 --conf-file job/flume-netcat-logger.conf -Dflume.root.logger=INFO,console
（7）使用netcat工具向本机的4444端口发送内容
nc localhost 44444
（8）Flume监听页面结果
实时监控Hive日志，并上传到HDFS中
（1）创建符合条件的flume配置文件；
（2）执行配置文件，开启监控
（3）开启Hive，生成日志
（4）查看HDFS上数据
（1）配置文件编写
要想读取 Linux 系统中的文件，就得按照 Linux 命令的规则执行命令。由于 Hive 日志在 Linux 系统中所以读取文件的类型选择：exec 即 execute 执行的意思。表示执行 Linux 命令来读取文件。
vim flume-file-hdfs.conf
# Name the components on this agent
a2.sources = r2
a2.sinks = k2
a2.channels = c2
# Describe/configure the source
a2.sources.r2.type = exec
a2.sources.r2.command = tail -F /opt/module/hive/logs/hive.log
# Describe the sink
a2.sinks.k2.type = hdfs
a2.sinks.k2.hdfs.path = hdfs://hadoop102:9820/flume/%Y%m%d/%H
#上传文件的前缀
a2.sinks.k2.hdfs.filePrefix = logs-
#是否按照时间滚动文件夹
a2.sinks.k2.hdfs.round = true
#多少时间单位创建一个新的文件夹
a2.sinks.k2.hdfs.roundValue = 1
#重新定义时间单位
a2.sinks.k2.hdfs.roundUnit = hour
#是否使用本地时间戳
a2.sinks.k2.hdfs.useLocalTimeStamp = true
#积攒多少个 Event 才 flush 到 HDFS 一次
a2.sinks.k2.hdfs.batchSize = 100
#设置文件类型，可支持压缩
a2.sinks.k2.hdfs.fileType = DataStream
#多久生成一个新的文件
a2.sinks.k2.hdfs.rollInterval = 60
#设置每个文件的滚动大小
a2.sinks.k2.hdfs.rollSize = 134217700
#文件的滚动与 Event 数量无关
a2.sinks.k2.hdfs.rollCount = 0
# Use a channel which buffers events in memory
a2.channels.c2.type = memory
a2.channels.c2.capacity = 1000
a2.channels.c2.transactionCapacity = 100
# Bind the source and sink to the channel
a2.sources.r2.channels = c2
a2.sinks.k2.channel = c2
（2）HDFS创建Flume监控目录
hadoop fs -mkdir -p /flume
（3）运行Flume
bin/flume-ng agent --conf conf/ --name a2 --conf-file job/flume-file-hdfs.conf
（4）开启 Hadoop 和 Hive 并操作 Hive 产生日志
bin/hive
（5）查看HDFS上结果
使用Flume监听整个目录的文件，并上传至HDFS
（1）创建符合条件的flume配置文件
（2）执行配置文件，开启监控
（3）向upload目录中添加文件
（4）查看HDFS上数据
（5）查看/opt/module/flume/upload目录上传的文件是否已经标记为.COMPLETED结尾；.tmp后缀结尾文件没有上传。
（1）创建配置文件
vim flume-dir-hdfs.conf
a3.sources = r3
a3.sinks = k3
a3.channels = c3
# Describe/configure the source
a3.sources.r3.type = spooldir
a3.sources.r3.spoolDir = /opt/module/flume/upload
a3.sources.r3.fileSuffix = .COMPLETED
a3.sources.r3.fileHeader = true
#忽略所有以.tmp 结尾的文件，不上传
a3.sources.r3.ignorePattern = ([^ ]*\.tmp)
# Describe the sink
a3.sinks.k3.type = hdfs
a3.sinks.k3.hdfs.path = hdfs://hadoop102:8020/flume/upload/%Y%m%d/%H
#上传文件的前缀
a3.sinks.k3.hdfs.filePrefix = upload-
#是否按照时间滚动文件夹
a3.sinks.k3.hdfs.round = true
#多少时间单位创建一个新的文件夹
a3.sinks.k3.hdfs.roundValue = 1
#重新定义时间单位
a3.sinks.k3.hdfs.roundUnit = hour
#是否使用本地时间戳
a3.sinks.k3.hdfs.useLocalTimeStamp = true
#积攒多少个 Event 才 flush 到 HDFS 一次
a3.sinks.k3.hdfs.batchSize = 100
#设置文件类型，可支持压缩
a3.sinks.k3.hdfs.fileType = DataStream
#多久生成一个新的文件
a3.sinks.k3.hdfs.rollInterval = 60
#设置每个文件的滚动大小大概是 128M
a3.sinks.k3.hdfs.rollSize = 134217700
#文件的滚动与 Event 数量无关
a3.sinks.k3.hdfs.rollCount = 0
# Use a channel which buffers events in memory
a3.channels.c3.type = memory
a3.channels.c3.capacity = 1000
a3.channels.c3.transactionCapacity = 100
# Bind the source and sink to the channel
a3.sources.r3.channels = c3
a3.sinks.k3.channel = c3
（2）启动监控文件夹命令
bin/flume-ng agent --conf conf/ --name a3 --conf-file job/flume-dir-hdfs.conf
（3）创建目录并添加文件
mkdir upload
touch atguigu.txt
touch atguigu.tmp
touch atguigu.log
（4）查看HDFS上的数据
使用Flume监听整个目录的实时追加文件，并上传至HDFS
（1）创建配置文件
vim flume-taildir-hdfs.conf
a3.sources = r3
a3.sinks = k3
a3.channels = c3
# Describe/configure the source
a3.sources.r3.type = TAILDIR
a3.sources.r3.positionFile = /opt/module/flume/tail_dir.json
a3.sources.r3.filegroups = f1 f2
a3.sources.r3.filegroups.f1 = /opt/module/flume/files/.*file.*
a3.sources.r3.filegroups.f2 = /opt/module/flume/files2/.*log.*
# Describe the sink
a3.sinks.k3.type = hdfs
a3.sinks.k3.hdfs.path =
hdfs://hadoop102:9820/flume/upload2/%Y%m%d/%H
#上传文件的前缀
a3.sinks.k3.hdfs.filePrefix = upload-
#是否按照时间滚动文件夹
a3.sinks.k3.hdfs.round = true
#多少时间单位创建一个新的文件夹
a3.sinks.k3.hdfs.roundValue = 1
#重新定义时间单位
a3.sinks.k3.hdfs.roundUnit = hour
#是否使用本地时间戳
a3.sinks.k3.hdfs.useLocalTimeStamp = true
#积攒多少个 Event 才 flush 到 HDFS 一次
a3.sinks.k3.hdfs.batchSize = 100
#设置文件类型，可支持压缩
a3.sinks.k3.hdfs.fileType = DataStream
#多久生成一个新的文件
a3.sinks.k3.hdfs.rollInterval = 60
#设置每个文件的滚动大小大概是 128M
a3.sinks.k3.hdfs.rollSize = 134217700
#文件的滚动与 Event 数量无关
a3.sinks.k3.hdfs.rollCount = 0
# Use a channel which buffers events in memory
a3.channels.c3.type = memory
a3.channels.c3.capacity = 1000
a3.channels.c3.transactionCapacity = 100
# Bind the source and sink to the channel
a3.sources.r3.channels = c3
a3.sinks.k3.channel = c3
（2）启动监控文件夹命令
bin/flume-ng agent --conf conf/ --name a3 --conf-file job/flume-taildir-hdfs.conf
（3）向files文件夹中追加文件
mkdir files
cd files/
#向files文件夹中追加文件
echo hello >> file1.txt
echo atguigu >> file2.txt
（4）查看HDFS上的数据
（1）Source接收数据后将数据封装成Event对象。
（2）Event数据对象经过Inteceptor链对数据进行改造
（3）根据Channel selector确定Event要发送给那个Channel
（4）Channel通过SinkProcessor分发给对应Sink
（1）推送事务流程
（2）拉取事务流程
简单串联就是将多个Flume顺序串联起来，从最初的Source开始到最终的Sink传送的目的存储系统。
Flume支持将事件流向一个或多个目的地，即1对多模式。
（1）负载均衡
将多个sink逻辑上分为一个sink组，sink组配合不同的SinkProcessor将数据相对均匀的分发到指定目录或者其他agent实例。
a1.sinkgroups.g1.processor.type =load_balance
a1.sinkgroups.g1.processor.backoff=true
（2）故障转移
有主备agent，主agent负责数据的采集、传输、落地，备用agent一直处于监听状态，一旦主agent宕机，备用agent启动，进行主agent的工作，直到主agent恢复。
每台服务器部署一个 flume 采集日志，传送到一个集中收集日志的flume，再由此 flume 上传到 hdfs、hive、hbase 等，进行日志分析。
nginx产生的日志数据需要供多个部门使用，如何处理一份数据发送两个系统来使用？
（1）创建目录
#创建文件夹
mkdir group1
cd group1
（2）创建flume1配置文件
#创建文件
vim flume-file-flume.conf
#文件内容如下
# Name the components on this agent
a1.sources = r1
a1.sinks = k1 k2
a1.channels = c1 c2
# 将数据流复制给所有 channel
a1.sources.r1.selector.type = replicating
# Describe/configure the source
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /opt/module/hive/logs/hive.log
a1.sources.r1.shell = /bin/bash -c
# Describe the sink
# sink 端的 avro 是一个数据发送者
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = hadoop102
a1.sinks.k1.port = 4141
a1.sinks.k2.type = avro
a1.sinks.k2.hostname = hadoop102
a1.sinks.k2.port = 4142
# Describe the channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
a1.channels.c2.type = memory
a1.channels.c2.capacity = 1000
a1.channels.c2.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1 c2
a1.sinks.k1.channel = c1
a1.sinks.k2.channel = c2
（3）创建Flume2的配置文件
vim flume-flume-hdfs.conf
#文件内容如下
# Name the components on this agent
a2.sources = r1
a2.sinks = k1
a2.channels = c1
# Describe/configure the source
# source 端的 avro 是一个数据接收服务
a2.sources.r1.type = avro
a2.sources.r1.bind = hadoop102
a2.sources.r1.port = 4141
# Describe the sink
a2.sinks.k1.type = hdfs
a2.sinks.k1.hdfs.path = hdfs://hadoop102:9820/flume2/%Y%m%d/%H
#上传文件的前缀
a2.sinks.k1.hdfs.filePrefix = flume2-
#是否按照时间滚动文件夹
a2.sinks.k1.hdfs.round = true
#多少时间单位创建一个新的文件夹
a2.sinks.k1.hdfs.roundValue = 1
#重新定义时间单位
a2.sinks.k1.hdfs.roundUnit = hour
#是否使用本地时间戳
a2.sinks.k1.hdfs.useLocalTimeStamp = true
#积攒多少个 Event 才 flush 到 HDFS 一次
a2.sinks.k1.hdfs.batchSize = 100
#设置文件类型，可支持压缩
a2.sinks.k1.hdfs.fileType = DataStream
#多久生成一个新的文件
a2.sinks.k1.hdfs.rollInterval = 30
#设置每个文件的滚动大小大概是 128M
a2.sinks.k1.hdfs.rollSize = 134217700
#文件的滚动与 Event 数量无关
a2.sinks.k1.hdfs.rollCount = 0
# Describe the channel
a2.channels.c1.type = memory
a2.channels.c1.capacity = 1000
a2.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a2.sources.r1.channels = c1
a2.sinks.k1.channel = c1
（4）创建flume3的配置文件
vim flume-flume-dir.conf
# Name the components on this agent
a3.sources = r1
a3.sinks = k1
a3.channels = c2
# Describe/configure the source
a3.sources.r1.type = avro
a3.sources.r1.bind = hadoop102
a3.sources.r1.port = 4142
# Describe the sink
a3.sinks.k1.type = file_roll
a3.sinks.k1.sink.directory = /opt/module/data/flume3
# Describe the channel
a3.channels.c2.type = memory
a3.channels.c2.capacity = 1000
a3.channels.c2.transactionCapacity = 100
# Bind the source and sink to the channel
a3.sources.r1.channels = c2
a3.sinks.k1.channel = c2
（5）分别执行配置文件
bin/flume-ng agent --conf conf/ --name a3 --conf-file job/group1/flume-flume-dir.conf
bin/flume-ng agent --conf conf/ --name a2 --conf-file job/group1/flume-flume-hdfs.conf
bin/flume-ng agent --conf conf/ --name a1 --conf-file job/group1/flume-file-flume.conf
（5）启动Hive后检查文件
使用Flume1监控一个端口，其sink组中的sink分别对接Flume2和Flume3，采用FailoverSinkProcessor，实现故障转移功能。
（1）创建目录
mkdir group2
cd group2
（2）创建flume1配置文件
vim flume-netcat-flume.conf
# Name the components on this agent
a1.sources = r1
a1.channels = c1
a1.sinkgroups = g1
a1.sinks = k1 k2
# Describe/configure the source
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44444
a1.sinkgroups.g1.processor.type = failover
a1.sinkgroups.g1.processor.priority.k1 = 5
a1.sinkgroups.g1.processor.priority.k2 = 10
a1.sinkgroups.g1.processor.maxpenalty = 10000
# Describe the sink
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = hadoop102
a1.sinks.k1.port = 4141
a1.sinks.k2.type = avro
a1.sinks.k2.hostname = hadoop102
a1.sinks.k2.port = 4142
# Describe the channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinkgroups.g1.sinks = k1 k2
a1.sinks.k1.channel = c1
a1.sinks.k2.channel = c1
（3）创建Flume2配置文件
vim flume-flume-console1.conf
# Name the components on this agent
a2.sources = r1
a2.sinks = k1
a2.channels = c1
# Describe/configure the source
a2.sources.r1.type = avro
a2.sources.r1.bind = hadoop102
a2.sources.r1.port = 4141
# Describe the sink
a2.sinks.k1.type = logger
# Describe the channel
a2.channels.c1.type = memory
a2.channels.c1.capacity = 1000
a2.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a2.sources.r1.channels = c1
a2.sinks.k1.channel = c1
（4）创建flume3配置文件
vim flume-flume-console2.conf
# Name the components on this agent
a3.sources = r1
a3.sinks = k1
a3.channels = c2
# Describe/configure the source
a3.sources.r1.type = avro
a3.sources.r1.bind = hadoop102
a3.sources.r1.port = 4142
# Describe the sink
a3.sinks.k1.type = logger
# Describe the channel
a3.channels.c2.type = memory
a3.channels.c2.capacity = 1000
a3.channels.c2.transactionCapacity = 100
# Bind the source and sink to the channel
a3.sources.r1.channels = c2
a3.sinks.k1.channel = c2
（5）分别执行相关配置文件
bin/flume-ng agent --conf conf/ --name a3 --conf-file job/group2/flume-flume-console2.conf -Dflume.root.logger=INFO,console
bin/flume-ng agent --conf conf/ --name a2 --conf-file job/group2/flume-flume-console1.conf -Dflume.root.logger=INFO,console
bin/flume-ng agent --conf conf/ --name a1 --conf-file job/group2/flume-netcat-flume.conf
（6）使用netcat工具向本机的44444端口发送内容
nc localhost 44444
#端口被占用
#1.查看端口是否被占用
netstat -anp |grep [端口号]
#2.查看占用的进程
lsof -i:[端口号]
#3.关闭进程
kill -9 [进程PID]
（7）正常工作
作用：在flume-console2正常工作时间，netcat内容只会发送到flume-console2中。
发送内容
发送结果：flume-console2打印结果
（8）故障转移
#杀掉flume-console2的进程
kill -9 pid
需求将来自不同服务器的日志数据采集回来存储在HDFS聚集到一起进行处理
待后续
（1）上游的Flume进程数十分庞大，直接送往目的地，目的地系统可能承载不了高并发的访问压力宕机；
（2）上游的Flume将采集完的数据送往下游Flume后，下游Flume进程将数据送往目的地；
（3）下游Flume数可人为控制，一般都有备份。
（1）Source：默认有的Avro（监听端口），thrift、Exec（执行linux命令）、JMS、SpoolingDirctory（监听目录）、第三方插件kafka；
（2）拦截器：所有events，增加头，类似json‘格式里的headers:{“keys”:“values”}:时间戳(头部插入时间戳),主机(头部插入主机名和IP)，静态(头部插入指定KV)，正则过滤(留下符合条件的)
（3）Channel：包括Memory、File、JDBC、Kafka等类型；
（4）拦截器：自定义拦截器，同上
（5）Sink：包括HDFS,hive,Avro,Hbase,kafka等类型
（1）负载均衡
flumeNG通过设置sinkgroups将多个沉潜节点分到一组，然后设置该组启用负载均衡，沉潜时会自动选择节点，如果节点宕机可选择其他节点:进程在后台运行时轮询依次查询,每个sink送出去的压力基本保持一致。
（2）事务机制
基于事务传输event(批量传输)，使用两个独立的事务分别处理source到channel和channel到sink之间，失败时会将所有数据都回滚到source或channel进行重试。
（1）数据丢失场景
（2）数据重复
总的来说，Flume 不会丢失数据，但是有可能造成数据的重复，例如数据已经成功由 Sink 发出， 但是没有接收到响应，Sink 会再次发送数据，此时可能会导致数据的重复
（1）memory：读写速度快,但是存储数据量小，Flume进程挂掉、服务器宕机或者重启都会导致数据丢失；
（2）File：落地到磁盘，如果sink已经提交完成的事务，则可以删除file；
（3）Kafka
日志收集层：只配置Source组件和kafka组件，不需要再配置Sink组件
日志汇聚层：只配置kafkachannel和sink，不需要再配置Source
（1）作用
（2）拦截器类型：ETL拦截器、区分类型拦截器
（3）自定义拦截器步骤
负载均衡和故障转移方案：配置sink组，同一个人sink组内有多个子sink，不同sink之间可以配置成负载均衡或故障转移。
（1）Flume内存配置为4G：实际开发中,再flume-env.sh中设置JVMheap为4G或更高
（2）FileChannel优化：DataDirs指向多个路径，每个路径对应不同硬盘，增大Flume的吞吐量:Checkpoint和backCheckpointDir也尽量配置在不同硬盘对应目录，保证checkpoint坏掉后，可快速恢复
（3）Sink优化：HDFSSink小文件处理