flume的原理和使用

sunren

　　flume是cloudera公司的一款高性能、高可能的分布式日志收集系统。
　　flume的核心是把数据从数据源收集过来，再送到目的地。为了保证输送一定成功，在送到目的地之前，会先缓存数据，待数据真正到达目的地后，删除自己缓存的数据。
　　flume传输的数据的基本单位是event，如果是文本文件，通常是一行记录，这也是事务的基本单位。
　　flume运行的核心是agent。它是一个完整的数据收集工具，含有三个核心组件，分别是source、channel、sink。通过这些组件，event可以从一个地方流向另一个地方，如下图所示。

　　source可以接收外部源发送过来的数据。不同的source，可以接受不同的数据格式。比如有目录池(spooling directory)数据源，可以监控指定文件夹中的新文件变化，如果目录中有文件产生，就会立刻读取其内容。
　　channel是一个存储地，接收source的输出，直到有sink消费掉channel中的数据。channel中的数据直到进入到下一个channel中或者进入终端才会被删除。当sink写入失败后，可以自动重启，不会造成数据丢失，因此很可靠。
　　sink会消费channel中的数据，然后送给外部源或者其他source。如数据可以写入到HDFS或者HBase中。
　　flume允许多个agent连在一起，形成前后相连的多级跳。
　　

使用

• 
  
• 1.从官网下载apache-flume-1.4.0-bin.tar.gz和apache-flume-1.4.0-src.tar.gz
  

　　2.分别解压，然后把src项目中的所有内容复制到bin项目中
　　3.删除src项目，把bin项目重命名为flume
　　4.配置到环境变量中
　　5.书写agent配置
　　使用flume的核心是如何配置agent文件。agent的配置是一个普通文本文件，使用键值对形式存储配置信息，可以设置多个agent信息。配置的内容包括source、channel、sink等。组件source、channel、sink都有名称、类型和很多个性化的属性配置。
　　配置文件应该这么写
　　# list the sources, sinks and channels for the agent
　　<Agent>.sources = <Source>
　　<Agent>.sinks = <Sink>
　　<Agent>.channels = <Channel1> <Channel2>
　　# set channel for source
　　<Agent>.sources.<Source>.channels = <Channel1> <Channel2> ...
　　# set channel for sink
　　<Agent>.sinks.<Sink>.channel = <Channel1>
　　# properties for sources
　　<Agent>.sources.<Source>.<someProperty> = <someValue>
　　# properties for channels
　　<Agent>.channel.<Channel>.<someProperty> = <someValue>
　　# properties for sinks
　　<Agent>.sources.<Sink>.<someProperty> = <someValue>
　　
　　# 下面是示例
　　#下面的agent1是代理名称，对应有source，名称是src1,；有一个sink，名称是sink1；有一个channel，名称是ch1.
　　agent1.sources = src1
　　agent1.sinks = sink1
　　agent1.channels = ch2
　　# 配置目录 source，监控目录(必须存在)的变化，要求文件名必须唯一，否则flume报错
　　agent1.sources.src1.type = spooldir
　　agent1.sources.src1.channels = ch2
　　agent1.sources.src1.spoolDir = /root/hmbbs
　　agent1.sources.src1.fileHeader = false
　　agent1.sources.src1.interceptors = i1
　　agent1.sources.src1.interceptors.i1.type = timestamp
　　# 配置内存 channel
　　agent1.channels.ch1.type = memory
　　agent1.channels.ch1.capacity = 1000
　　agent1.channels.ch1.transactionCapacity = 1000
　　agent1.channels.ch1.byteCapacityBufferPercentage = 20
　　agent1.channels.ch1.byteCapacity = 800000
　　# 配置文件 channel
　　agent1.channels.ch2.type = file
　　agent1.channels.ch2.checkpointDir = /root/flumechannel/checkpoint
　　agent1.channels.ch2.dataDirs = /root/flumechannel/data
　　# 配置hdfs sink
　　agent1.sinks.sink1.type = hdfs
　　agent1.sinks.sink1.channel = ch2
　　agent1.sinks.sink1.hdfs.path = hdfs://hadoop0:9000/flume/%Y-%m-%d/
　　agent1.sinks.sink1.hdfs.rollInterval=1
　　agent1.sinks.sink1.hdfs.fileType = DataStream
　　agent1.sinks.sink1.hdfs.writeFormat = Text
　　# 配置hbase sink
　　#配置hbase sink2
　　agent1.sinks.sink2.type = hbase
　　agent1.sinks.sink2.channel = channel1
　　agent1.sinks.sink2.table = hmbbs
　　agent1.sinks.sink2.columnFamily = cf
　　agent1.sinks.sink2.serializer = flume.HmbbsHbaseEventSerializer
　　agent1.sinks.sink2.serializer.suffix = timestamp
　　agent1.sinks.sink2.serializer = org.apache.flume.sink.hbase.SimpleHbaseEventSerializer
　　5.启动代理的脚本是flume-ng agent，需要指定agent name、配置目录、配置文件
　　-n 指定agent名称
　　-c 指定配置文件目录
　　-f 指定配置文件
　　-Dflume.root.logger=DEBUG,console
　　因此完整的启动命令应该这么写
　　bin/flume-ng agent –n agent1 –c conf –f conf/example –Dflume.root.logger=DEBUG,console
　　启动成功后，可以向目录/root/hmbbs中放入文件，flume会感知到新文件，然后上传到hdfs的/flume目录下。