flume集群日志收集

小程序：扫一扫查出行
【扫一扫了解最新限行尾号】
复制小程序

一、Flume简介

Flume是一个分布式的、高可用的海量日志收集、聚合和传输日志收集系统，支持在日志系统中定制各类数据发送方（如：Kafka，HDFS等），便于收集数据。其核心为agent，agent是一个java进程，运行在日志收集节点。

agent里面包含3个核心组件：source、channel、sink。
source组件是专用于收集日志的，可以处理各种类型各种格式的日志数据,包括avro、thrift、exec、jms、spooling directory、netcat、sequence generator、syslog、http、legacy、自定义，同时 source组件把数据收集

以后，临时存放在channel中。

channel组件是在agent中专用于临时存储数据的，可以存放在memory、jdbc、file、自定义等。channel中的数据只有在sink发送成功之后才会被删除。

sink组件是用于把数据发送到目的地的组件，目的地包括hdfs、logger、avro、thrift、ipc、file、null、hbase、solr、自定义。

在整个数据传输过程中，流动的是event。事务保证是在event级别。flume可以支持多级flume的agent，支持扇入(fan-in)、扇出(fan-out)。

二、环境准备

1）hadoop集群（楼主用的版本2.7.3，共6个节点，可参考http://www.cnblogs.com/qq503665965/p/6790580.html）

2）flume集群规划：

HOST	作用	方式	路径
hadoop01	agent	spooldir	/home/hadoop/logs
hadoop05	collector	HDFS	/logs
hadoop06	collector	HDFS	/logs

其基本结构官网给出了更加具体的说明，这里楼主就直接copy过来了：

三、集群配置

1）系统环境变量配置

export FLUME_HOME=/home/hadoop/apache-flume-1.7.0-bin
export PATH=$PATH:$JAVA_HOME/bin:$HADOOP_HOME/bin:$HADOOP_HOME/sbin:$FLUME_HOME/bin

记得 source /etc/profile 。

2）flume JDK环境

mv flume-env.sh.templete flume-env.sh
vim flume-env.sh
export JAVA_HOME=/usr/jdk1.7.0_60//增加JDK安装路径即可

3）hadoop01中flume配置

在conf目录增加配置文件 flume-client ,内容为：

#agent1名称
agent1.channels = c1
agent1.sources = r1
agent1.sinks = k1 k2 #sink组名称
agent1.sinkgroups = g1 #set channel
agent1.channels.c1.type = memory
agent1.channels.c1.capacity = 1000
agent1.channels.c1.transactionCapacity = 100 agent1.sources.r1.channels = c1
agent1.sources.r1.type = spooldir
#日志源
agent1.sources.r1.spoolDir =/home/hadoop/logs agent1.sources.r1.interceptors = i1 i2
agent1.sources.r1.interceptors.i1.type = static
agent1.sources.r1.interceptors.i1.key = Type
agent1.sources.r1.interceptors.i1.value = LOGIN
agent1.sources.r1.interceptors.i2.type = timestamp # 设置sink1
agent1.sinks.k1.channel = c1
agent1.sinks.k1.type = avro
#sink1所在主机
agent1.sinks.k1.hostname = hadoop05
agent1.sinks.k1.port = 52020 #设置sink2
agent1.sinks.k2.channel = c1
agent1.sinks.k2.type = avro
#sink2所在主机
agent1.sinks.k2.hostname = hadoop06
agent1.sinks.k2.port = 52020 #设置sink组包含sink1，sink2
agent1.sinkgroups.g1.sinks = k1 k2 #高可靠性
agent1.sinkgroups.g1.processor.type = failover
#设置优先级
agent1.sinkgroups.g1.processor.priority.k1 = 10
agent1.sinkgroups.g1.processor.priority.k2 = 1
agent1.sinkgroups.g1.processor.maxpenalty = 10000

4）hadoop05配置

#设置 Agent名称
a1.sources = r1
a1.channels = c1
a1.sinks = k1 #设置channlels
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100 # 当前节点信息
a1.sources.r1.type = avro
#绑定主机名
a1.sources.r1.bind = hadoop05
a1.sources.r1.port = 52020
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = static
a1.sources.r1.interceptors.i1.key = Collector
a1.sources.r1.interceptors.i1.value = hadoop05
a1.sources.r1.channels = c1 #sink的hdfs地址
a1.sinks.k1.type=hdfs
a1.sinks.k1.hdfs.path=/logs
a1.sinks.k1.hdfs.fileType=DataStream
a1.sinks.k1.hdfs.writeFormat=TEXT
#没1K产生文件
a1.sinks.k1.hdfs.rollInterval=1
a1.sinks.k1.channel=c1
#文件后缀
a1.sinks.k1.hdfs.filePrefix=%Y-%m-%d

5）hadoop06配置

#设置 Agent名称
a1.sources = r1
a1.channels = c1
a1.sinks = k1 #设置channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100 # 当前节点信息
a1.sources.r1.type = avro
#绑定主机名
a1.sources.r1.bind = hadoop06
a1.sources.r1.port = 52020
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = static
a1.sources.r1.interceptors.i1.key = Collector
a1.sources.r1.interceptors.i1.value = hadoop06
a1.sources.r1.channels = c1
#设置sink的hdfs地址目录
a1.sinks.k1.type=hdfs
a1.sinks.k1.hdfs.path=/logs
a1.sinks.k1.hdfs.fileType=DataStream
a1.sinks.k1.hdfs.writeFormat=TEXT
a1.sinks.k1.hdfs.rollInterval=1
a1.sinks.k1.channel=c1
a1.sinks.k1.hdfs.filePrefix=%Y-%m-%d