Linux环境Flume安装配置及使用

1. 认识Flume

(1) Flume介绍

• 日志收集系统
• 官网：http://flume.apache.org/
• 概述：Flume是一种分布式，可靠且可用的服务，用于有效地收集，聚合和移动大量日志数据。它具有基于流数据流的简单灵活的架构。它具有可靠的可靠性机制和许多故障转移和恢复机制，具有强大的容错性。它使用简单的可扩展数据模型，允许在线分析应用程序。
• 大数据阶段：
  • <1>. 数据采集(爬虫/日志数据/flume)
  • <2>. 数据存储(hdfs/hive/hbase(nosql))
  • <3>. 数据计算(mapreduce/hive/sparkSQL/sparkStreaming/flink)
  • <4>. 数据可视化(echart/quickBI)

(2) Flume角色

• <1>. source
  • 数据源，用户从数据发生器采集接收数据，source产生数据流，同时会把产生的数据流以Flume的event格式传输到一个或者多个channel。
• <2>. channel
  • 传输通道，短暂的存储容器，将从source处接收到的event格式的数据以队列形式缓存起来，直到它们被sinks消费掉，它在source和sink间起桥梁的作用，channal是一个完整的事务，这一点保证了数据在收发的时候的一致性. 并且它可以和任意数量的source和sink链接。
• <3>. sink
  • 下沉，用于消费channel传输的数据，将数据源传递到目标源，目标可能是另一个sink，也可能HDFS、HBase，最终将数据存储到集中存储器。
• <4>. event
  • 在flume中使用事件作为传输的基本单元。

(3) Flume使用

• 简单易用，只需要写配置文件即可。

2. Flume-1.6.0安装配置流程

(1) Flume环境前提：

• Java运行环境

(2) 解压apache-flume-1.6.0-bin.tar.gz安装包到目标目录下：

• tar -zxvf .tar.gz -C 目标目录

(3) 为后续方便，重命名Flume文件夹：

• mv apache-flume-1.6.0-bin/ flume-1.6.0

(4) 修改配置文件：

• 进入flume-1.6.0/conf路径，重命名配置文件：
  • mv flume-env.sh.template flume-env.sh
• 修改flume-env.sh信息：
  • vi flume-env.sh

  • # Enviroment variables can be set here.export JAVA_HOME=jdk路径复制代码

(5) 配置环境变量：

• 修改配置文件：
  • vi /etc/profile
• 增加以下内容：
  • export FLUME_HOME=flume安装路径
  • export PATH=$PATH:$FLUME_HOME/bin
• 声明环境变量：
  • source /etc/profile

(6) 启动

• flume-ng agent 使用ng启动agent --conf YYYY/ 指定配置所在的文件夹 --name a1指定的agent别名 --conf-file YYYY/XXXXXX 指定配置文件 -Dflume.root.logger=INFO,console 可选，指定日志输出级别(输出到控制台) & 可选，Flume在后台运行
• 旧版本：flume-og ——在bin目录下查看

3. Flume监听端口

(1) 编辑配置文件：

• 在flume/conf目录下，创建配置文件flumejob_telnet.conf：
  • vi flumejob_telnet.conf

  • # Flume监听端口——配置文件# Name the components on this agent  定义变量方便调用 加s可以有多个此角色a1.sources = r1a1.sinks = k1a1.channels = c1# Describe/configure the source  描述source角色 进行内容定制# 此配置属于tcp source 必须是netcat类型a1.sources.r1.type = netcata1.sources.r1.bind = localhosta1.sources.r1.port = 44444# Describe the sink  输出日志文件a1.sinks.k1.type = logger# Use a channel which buffers events in memory（file） 使用内存 总大小1000 每次传输100a1.channels.c1.type = memorya1.channels.c1.capacity = 1000a1.channels.c1.transactionCapacity = 100# Bind the source and sink to the channel  一个source可以绑定多个channel# 一个sink绑定一个channela1.sources.r1.channels = c1a1.sinks.k1.channel = c1复制代码
• 保存并退出：
  • :wq

(2) NetCat

• 介绍：netcat是网络工具中的“瑞士军刀”，它能通过TCP和UDP在网络中读写数据。通过与其他工具结合和重定向，你可以在脚本中以多种方式使用它。netcat所做的就是在两台电脑之间建立链接并返回两个数据流。
• 安装：
  • yum install nc

(3) Telnet

• 介绍：telnet协议是TCP/IP协议族中的一员，是Internet远程登录服务的标准协议和主要方式。它为用户提供了在本地计算机上完成远程主机工作的能力。在终端使用者的电脑上使用telnet程序，用它连接到服务器。终端使用者可以在telnet程序中输入命令，这些命令会在服务器上运行，就像直接在服务器的控制台上输入一样。可以在本地就能控制服务器。
• 安装：
  • yum install telnet.x86_64

(4) 运行：

• 启动Flume：
  • flume-ng agent --conf conf/ --name a1 --conf-file conf/flumejob_telnet.conf -Dflume.root.logger=INFO,console
• 打开客户端副本：
  • telnet localhost 44444
• 在副本输入数据，客户端会收到相关监听信息

4. Flume监听本地Linux-hive日志文件采集到HDFS

(1) 编辑配置文件：

• 在flume/conf目录下，创建配置文件flumejob_hdfs.conf：
  • vi flumejob_hdfs.conf

  • # Flume监听本地Linux-hive日志文件采集到HDFS——配置文件# Name the components on this agent  agent别名设置a1.sources = r1a1.sinks = k1a1.channels = c1# Describe/configure the source  设置数据源监听本地文件配置# exec 执行一个命令的方式去查看文件 tail -F 实时查看a1.sources.r1.type = exec# 要执行的脚本command tail -F 默认10行 man tail  查看帮助# 监听hive操作日志a1.sources.r1.command = tail -F /tmp/root/hive.log# 执行这个command使用的是哪个脚本 -c 指定使用什么命令# whereis bash# bash: /usr/bin/bash /usr/share/man/man1/bash.1.gza1.sources.r1.shell = /usr/bin/bash -c# Describe the sink  设置sink# 指定sink类型a1.sinks.k1.type = hdfs# 指定HDFS路径 %Y%m%d/%H%M%S 日期时间  ————修改项a1.sinks.k1.hdfs.path = hdfs://bigdata01:9000/flume/%Y%m%d/%H-%M#上传文件的前缀a1.sinks.k1.hdfs.filePrefix = logs-#是否按照时间滚动文件夹a1.sinks.k1.hdfs.round = true#多少时间单位创建一个新的文件夹  秒 （默认30s）a1.sinks.k1.hdfs.roundValue = 1#重新定义时间单位（每分钟滚动一个文件夹）a1.sinks.k1.hdfs.roundUnit = minute#是否使用本地时间戳a1.sinks.k1.hdfs.useLocalTimeStamp = true#积攒多少个 Event 才 flush 到 HDFS 一次a1.sinks.k1.hdfs.batchSize = 500#设置文件类型，可支持压缩a1.sinks.k1.hdfs.fileType = DataStream#多久生成一个新的文件 秒a1.sinks.k1.hdfs.rollInterval = 30#设置每个文件的滚动大小 字节（最好128M）a1.sinks.k1.hdfs.rollSize = 134217700#文件的滚动与 Event 数量无关a1.sinks.k1.hdfs.rollCount = 0#最小冗余数(备份数 生成滚动功能则生效roll hadoop本身有此功能 无需配置) 1份 不冗余a1.sinks.k1.hdfs.minBlockReplicas = 1# Use a channel which buffers events in memory  设置channel  使用内存 总大小1000 每次传输100a1.channels.c1.type = memorya1.channels.c1.capacity = 1000a1.channels.c1.transactionCapacity = 100# Bind the source and sink to the channel  指定channela1.sources.r1.channels = c1a1.sinks.k1.channel = c1复制代码
• 保存并退出：
  • :wq

(2) Flume+Hive

• 把Hive相关Hadoop依赖包导入Flume：
  • 进入Flume包路径：
    • cd /XXXX/flume/lib
  • 上传相关jar包

(3) 运行：

• 启动Flume：
  • flume-ng agent --conf conf/ --name a1 --conf-file conf/flumejob_hdfs.conf
• 之后操作Hive，相关日志文件上传至hdfs指定目录

5. Flume监听本地文件夹

(1) 编辑配置文件：

• 在flume/conf目录下，创建配置文件flumejob_dir.conf：
  • vi flumejob_dir.conf

  • # Flume监听文件夹# Name the components on this agent  agent别名设置a1.sources = r1a1.sinks = k1a1.channels = c1# Describe/configure the source  设置数据源监听本地文件配置a1.sources.r1.type = spooldir# 监控的文件夹a1.sources.r1.spoolDir = /root/testdir# 上传成功后显示后缀名a1.sources.r1.fileSuffix = .COMPLETED# 加绝对路径的文件名 默认为falsea1.sources.r1.fileHeader = true# 忽略所有以.tmp结尾的文件（正在被写入）# ^以任何开头 出现无限次 以.tmp结尾的文件a1.sources.r1.ignorePattern = ([^ ]*\.tmp)# Describe the sink  设置sink 下沉到hdfs# 指定sink类型a1.sinks.k1.type = hdfs# 指定HDFS路径 %Y%m%d/%H%M%S 日期时间  ————修改项a1.sinks.k1.hdfs.path = hdfs://bigdata01:9000/flume/testdir/%Y%m%d/%H-%M# 上传文件的前缀a1.sinks.k1.hdfs.filePrefix = testdir-#是否按照时间滚动文件夹a1.sinks.k1.hdfs.round = true#多少时间单位创建一个新的文件夹 （默认30s）a1.sinks.k1.hdfs.roundValue = 1#重新定义时间单位a1.sinks.k1.hdfs.roundUnit = hour#是否使用本地时间戳a1.sinks.k1.hdfs.useLocalTimeStamp = true#积攒多少个 Event 才 flush 到 HDFS 一次a1.sinks.k1.hdfs.batchSize = 100#设置文件类型，可支持压缩a1.sinks.k1.hdfs.fileType = DataStream#多久生成一个新的文件 秒a1.sinks.k1.hdfs.rollInterval = 600#设置每个文件的滚动大小 字节（最好128M）a1.sinks.k1.hdfs.rollSize = 134217700#文件的滚动与 Event 数量无关a1.sinks.k1.hdfs.rollCount = 0#最小冗余数(备份数 生成滚动功能则生效roll hadoop本身有此功能 无需配置) 1份 不冗余a1.sinks.k1.hdfs.minBlockReplicas = 1# Use a channel which buffers events in memory  设置channel  使用内存 总大小1000 每次传输100a1.channels.c1.type = memorya1.channels.c1.capacity = 1000a1.channels.c1.transactionCapacity = 100# Bind the source and sink to the channel  指定channela1.sources.r1.channels = c1a1.sinks.k1.channel = c1复制代码
• 保存并退出：
  • :wq

(2) 运行：

• 启动Flume：
  • flume-ng agent --conf conf/ --name a1 --conf-file conf/flumejob_dir.conf
• 之后对指定的文件夹进行操作，相关文件夹文件信息上传至hdfs指定目录。
• 注意文件夹内只能有文件不能有目录。

6. Flume多channel结构配置

(1) 实现目标

• 获取数据源后由Flume-a1进行处理，分两个端口发送数据，Flume-a2接收到数据后经过处理下沉到hdfs，Flume-a3接收到数据后经过处理下沉到本地。
• 流程图如下：

(2) 编辑配置文件1：

• 在flume/conf目录下，创建配置文件flumejob_a1.conf：
  • vi flumejob_a1.conf

  • # Flume多channel结构配置a1# Name the components on this agent  agent别名设置a1.sources = r1a1.sinks = k1 k2a1.channels = c1 c2# 将数据流复制给多个channela1.sources.r1.selector.type = replicating# Describe/configure the source  设置数据源监听本地文件配置# exec 执行一个命令的方式去查看文件 tail -F 实时查看a1.sources.r1.type = exec# 要执行的脚本command tail -F 默认10行 man tail  查看帮助# 监听hive操作日志a1.sources.r1.command = tail -F /tmp/root/hive.log# 执行这个command使用的是哪个脚本 -c 指定使用什么命令# whereis bash# bash: /usr/bin/bash /usr/share/man/man1/bash.1.gza1.sources.r1.shell = /usr/bin/bash -c# Describe the sink  设置sink# 分两个端口发送数据a1.sinks.k1.type = avroa1.sinks.k1.hostname = bigdata01a1.sinks.k1.port = 4141a1.sinks.k2.type = avroa1.sinks.k2.hostname = bigdata01a1.sinks.k2.port = 4142# Use a channel which buffers events in memory  设置channel  使用内存 总大小1000 每次传输100a1.channels.c1.type = memorya1.channels.c1.capacity = 1000a1.channels.c1.transactionCapacity = 100a1.channels.c2.type = memorya1.channels.c2.capacity = 1000a1.channels.c2.transactionCapacity = 100# Bind the source and sink to the channel  指定channela1.sources.r1.channels = c1 c2a1.sinks.k1.channel = c1a1.sinks.k2.channel = c2复制代码
• 保存并退出：
  • :wq

(3) 编辑配置文件2：

• 在flume/conf目录下，创建配置文件flumejob_a2.conf：
  • vi flumejob_a2.conf

  • # Flume多channel结构配置a2# 接收a1数据下沉到hdfs# Name the components on this agent  agent别名设置a2.sources = r1a2.sinks = k1a2.channels = c1# Describe/configure the source  设置数据源监听本地文件配置a2.sources.r1.type = avro# 获取数据a2.sources.r1.bind = bigdata01a2.sources.r1.port = 4141# Describe the sink  设置sink# 指定sink类型a2.sinks.k1.type = hdfs# 指定HDFS路径 %Y%m%d/%H%M%S 日期时间  ————修改项a2.sinks.k1.hdfs.path = hdfs://bigdata01:9000/flume1/%Y%m%d/%H-%M# 上传文件的前缀a2.sinks.k1.hdfs.filePrefix = flume1-# 是否按照时间滚动文件夹a2.sinks.k1.hdfs.round = true# 多少时间单位创建一个新的文件夹 （默认30s）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 = 600# 设置每个文件的滚动大小 字节（最好128M）a2.sinks.k1.hdfs.rollSize = 134217700# 文件的滚动与 Event 数量无关a1.sinks.k1.hdfs.rollCount = 0# 最小冗余数(备份数 生成滚动功能则生效roll hadoop本身有此功能 无需配置) 1份 不冗余a2.sinks.k1.hdfs.minBlockReplicas = 1# Use a channel which buffers events in memory  设置channel  使用内存 总大小1000 每次传输100a2.channels.c1.type = memorya2.channels.c1.capacity = 1000a2.channels.c1.transactionCapacity = 100# Bind the source and sink to the channel  指定channela2.sources.r1.channels = c1a2.sinks.k1.channel = c1复制代码
• 保存并退出：
  • :wq

(4) 编辑配置文件3：

• 在flume/conf目录下，创建配置文件flumejob_a3.conf：
  • vi flumejob_a3.conf

  • # Flume多channel结构配置a3# 接收a1数据下沉到本地# Name the components on this agent  agent别名设置a3.sources = r1a3.sinks = k1a3.channels = c1# Describe/configure the source  设置数据源监听本地文件配置a3.sources.r1.type = avro# 获取数据a3.sources.r1.bind = bigdata01a3.sources.r1.port = 4142# Describe the sink  设置sink# 指定sink类型a3.sinks.k1.type = file_rolla3.sinks.k1.sink.directory = /root/flume1# Use a channel which buffers events in memory  设置channel  使用内存 总大小1000 每次传输100a3.channels.c1.type = memorya3.channels.c1.capacity = 1000a3.channels.c1.transactionCapacity = 100# Bind the source and sink to the channel  指定channela3.sources.r1.channels = c1a3.sinks.k1.channel = c1复制代码
• 保存并退出：
  • :wq
• **注意：**本地文件夹需要提前创建好

(5) 运行：

• 启动Flume-a1：
  • flume-ng agent --conf conf/ --name a1 --conf-file conf/flumejob_a1.conf
• 启动Flume-a2：
  • flume-ng agent --conf conf/ --name a2 --conf-file conf/flumejob_a2.conf
• 启动Flume-a3：
  • flume-ng agent --conf conf/ --name a3 --conf-file conf/flumejob_a3.conf
• a1最先启动，a2、a3先后启动顺序无限制。