Flume实践

What

Flume 海量日志收集框架

分布式
可扩展
可靠
高可用

0.9x 的架构 OG

单点故障问题,使用zookeeper高可用

1.0x 的架构 NG

Agent

由source、channel、sink三大组件组成，如图：

source 采集
- 从Client收集数据，传递给Channel。
- 可以接收外部源发送过来的数据。
- 不同的 source，可以接受不同的数据格式。
channel 存储池
- 接收source的输出
- 直到有sink消费掉channel中的数据Channel中的数据直到进入到下一个channel中或者进入终端才会被删除
- sink写入失败后，可以自动重启，不会造成数据丢失，因此很可靠。
sink 输出

数据库可靠性

end-to-end
store on failure
best effort

自身可扩展性

1.0自身agent实现扩展

功能可扩展性

Flume自带了很多组件，包括各种agent（file，syslog，HDFS等）

Why

Where

http://flume.apache.org/

avro简介

说明

http://flume.apache.org/FlumeUserGuide.html

How

安装

下载

配置flume-env.sh

1
2
3

cd /opt/soft/apache-flume-1.7.0-bin/conf
cp flume-env.sh.template flume-env.sh
vi flume-env.sh

修改java_home

配环境变量 vi ~/.base_profile 添加flume home
1
2
export FLUME_HOME=/opt/soft/apache-flume-1.7.0-bin
PATH=$PATH:$FLUME_HOME/bin
使立刻生效source ~/.bash_profile
安装telnet yum install telnet.x86_64

配置文件 netcat_logger.conf

# example.conf: A single-node Flume configuration

# 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

启动

1	flume-ng agent --conf conf --conf-file netcat_logger.conf --name a1 -Dflume.root.logger=INFO,console

需在 netcat_logger.conf 这个目录执行，否则需要全路径

测试，另开一个ssh 窗口启动 telnet telnet localhost 44444

模式

netcat_logger
netcat_hdfs

exec_hdfs

tailf *.log

# a1 which ones we want to activate.
 a1.channels = c1
 a1.sources = r1
 a1.sinks = k1

 # Describe/configure the source
 a1.sources.r1.type = exec
 a1.sources.r1.command = tail -F /root/flume.log

 a1.sinks.k1.type = hdfs
 a1.sinks.k1.hdfs.path = hdfs://appcity/flumelog/%y-%m-%d
 a1.sinks.k1.hdfs.rollCount=0
 a1.sinks.k1.hdfs.rollInterval=0
 a1.sinks.k1.hdfs.rollSize=10240
 a1.sinks.k1.hdfs.idleTimeout=5
 a1.sinks.k1.hdfs.fileType=DataStream
 a1.sinks.k1.hdfs.useLocalTimeStamp=true

 # Define a memory channel called c1 on a1
 a1.channels.c1.type = memory

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

avro_logger