一、Flume概述

      Flume是一个分布式的数据收集系统， 具有高可靠、 高可用、 事务管理、 失 败重启等功能。 数据处理速度快， 完全可以用于生产环境。

      Flume的核心是agent。 agent是一个java进程， 运行在日志收集端， 通过agent 接收日志， 然后暂存起来， 再发送到目的地。 

     

 

1、Flume是一个海量日志收集系统，由Cloudera公司推出。

2、Flume中，事务(Event)是最小的数据单位，通常代表一行记录（一条记

录）。Event由source从外部数据源获取，传给channel暂存，然后传

给sink，通过sink传到目的地(HDFS)中。

3、Flume中，Agent是运行的核心，它包括3个组件：

1）source，可以从外部数据源中获取数据。例如，它可以对某一

目录（文件夹）进行监控，如果目录中有新的文件产生，source会

自动读取新文件的内容，然后将内容传递给channel进行暂存。

2）channel，可以接收source传递的数据，并将数据进行暂存，

然后将数据传递给sink。直到sink将数据成功传到目的地之后，

channel才将暂存的数据删除。

channel是source与sink之间连接的桥梁，source与sink之间不能

直接发生连接，只能通过channel进行连接。

3）sink，可以接收channel传递的数据，并将数据传到目的地中。

 

二、Flume核心概念

• 核心概念： agent、 source、 channel、 sink、 interceptor

• source可以接收各种格式的数据， 如console、 thrift、 exec、 httpdent

• sink可以发送到各种不同的目的地， 如hdfs、 hbase、 文件、 控制台等

• channel缓冲数据， 可以放在内存、 文件、 数据库等

• interceptor可以在数据传递过程中改变其属性信息

三、安装步骤：

1、解压缩：tar -zxvf apache-flume-1.6.0-bin.tar.gz

重命名：mv ******* flume

配置相关的环境变量:

vi    .bash_profile

export FLUME_HOME=/home/hadoop/apps/flume

export FLUME_CONF_DIR=$FLUME_HOME/conf

export PATH=$PATH:$FLUME_HOME/bin

2、vi flume-conf.properties

#agent1表示代理名称

agent1.sources=source1

agent1.sinks=sink1

agent1.channels=channel1

• #配置source1

• agent1.sources.source1.type=spooldir

• agent1.sources.source1.spoolDir=/home/hadoop/flumeinfo

• agent1.sources.source1.channels=channel1

• agent1.sources.source1.fileHeader = false

• agent1.sources.source1.interceptors = i1

• agent1.sources.source1.interceptors.i1.type = timestamp

• #配置channel1

• agent1.channels.channel1.type=file

• agent1.channels.channel1.checkpointDir=/home/hadoop/flume_tmp_cp

• agent1.channels.channel1.dataDirs=/home/hadoop/flume_tmp

• #配置sink1

• agent1.sinks.sink1.type=hdfs

• agent1.sinks.sink1.hdfs.path=hdfs://liuwei1:9000/logs(如果高可就集群则hdfs://myha01/logs)

• agent1.sinks.sink1.hdfs.fileType=DataStream

• agent1.sinks.sink1.hdfs.writeFormat=TEXT

• agent1.sinks.sink1.hdfs.rollInterval=1

• agent1.sinks.sink1.channel=channel1

• agent1.sinks.sink1.hdfs.filePrefix=%Y-%m-%d

  

3、创建本地文件夹： mkdir /home/hadoop/flumeinfo                   mkdir /home/hadoop/flume_tmp_cp           mkdir /home/hadoop/flume_tmp

      创建HDFS文件夹： hdfs dfs -mkdir /logs

4、启动脚本

进入bin目录下    执行  ./flume-ng agent -n agent1 -c conf -f /home/hadoop/apps/flume/conf/flume-conf.properties -Dflume.root.logger=DEBUG,console

新建一份文件， 移动到/home/hadoop/flumeinfo目录下， flume就会自动上传到HDFS的/logs目录中

四、Flume+Spark Streaming 

vi flume-conf.properties1 

修改 配置sink1 （其他不用修改）

agent1.sinks.sink1.type = avro

agent1.sinks.sink1.channel = channel1

agent1.sinks.sink1.hostname =0.0.0.0

agent1.sinks.sink1.port = 41414

再启动  ./flume-ng agent -n agent1 -c conf -f /home/hadoop/apps/flume/conf/flume-conf.properties1 -Dflume.root.logger=DEBUG,console （启动相关flume-conf.properties1）

object produce {  def main(args: Array[String]): Unit = {    val conf=new SparkConf().setAppName("produce").setMaster("local[2]")    val ssc=new StreamingContext(conf,Seconds(5))    val linesDStream=FlumeUtils.createStream(ssc,"0.0.0.0",41414)    val wordsDStream=linesDStream.flatMap(line=>{      val inf=new String(line.event.getBody.array())      inf.split(" ")    })    val resultDStream=wordsDStream.map((_,1)).reduceByKey(_+_)    resultDStream.print()    ssc.start()    ssc.awaitTermination()  }