Flume学习笔记

1 简介

(1) Apache Flume是一个分布式、可信任的数据采集、日志收集弹性系统(框架)，用于高效收集、汇聚和移动大规模日志信息从多种不同的数据源到一个集中的数据存储中心(HDFS、Hbase或者本地文件系统)

(2) 可信任是指保证消息有效的处理和传递：

如果节点出现异常、导致数据传输过程中断，进行数据回滚，或者数据重发；
对于同一节点，source向channel写数据，是一个一个批次写的，如果该批次内的数据出现异常，则不会写入channel中，同批次其他正常数据也不会写入channel中(但是，如果已经接受到的部分数据，对这部分进行抛弃)，依靠上一节点重新发送数据。

(3) 数据来源：Console、RPC、Text、Tail、Syslog、Exec

(4) Flume接收到的数据可以输出到哪里？

磁盘、hdfs、hbase、网络传输给下游、Kafka

(5) Flume的核心：agent，一个完整的数据收集工具

(6) Flume中最基本的数据单元:Event

(7) Flume 基于流式架构，容错性强，也很灵活简单。

(8) Flume、 Kafka 用来实时进行数据收集，Spark、 Storm 用来实时处理数据， impala 用来实时查询。

(9) 数据处理流程：数据源---->数据采集---->缓冲器---->过滤---->转换---->存储---->索引--->分析

Data—>flume—>kafka—>storm—>(Streaming) —>hbase(nosql/httpapi)

2 特点

(1) Flume可以高效率的将多个网站服务器中收集的日志信息存入HDFS/Hbase中。

(2) 使用Flume，可以将从多个服务器中获取的数据迅速的移交给Hadoop中。

(3) 支持各种接入资源数据的类型以及接出数据类型。

(4) 可以被水平扩展和垂直扩展。

(5) 支持多路径流量，多管道接入流量，多管道接出流量，上下文路由等。

多管道接入即多个数据源，比如log1是收集展现日记，log2是收集点击日记，多管道接出即多个后端存储，存储后也可以是不同的存储方法。

(6)flume中的数据存储，两种方法：

复用机制：后端指定的存储节点才会存储指定的数据(一对一)
复制机制：后端的所有节点都会存储一份数据(一对多)

3 架构

(1) 数据发生器：即日志服务器，产生的数据被单个的运行在数据发生器所在服务器的agent收集，之后数据收容器从各个agent上汇集数据并将采集的数据存入到HDFS或者Hbase中

(2) agent：代理模块，对消息进行接收和汇集，与日记服务器同一节点

(3) data collector：接收agent之后的数据，处理之后再发送给后端存储

(4) agent与collector是在不同节点，解耦

(5) agent与server机器一比一，多个agent对应一个collector

4 Event

(1) 数据单位：storm(tuple)、HDFS(block)、Flume(Event事件)；

(2) Flume使用Event对象来作为传递数据的格式，是内部数据传输的最基本单元；

(3) Event有两部分：转载数据的字节数组(body)+可选头部(header)；

(4) Body：是一个字节数组，存储实际数据；

(5) Header：是key/value形式的，根据key给后端分发。相当于partition分桶；

可以用来制造路由决策或者携带其它结构化信息(如事件的时间戳或事件来源的服务器主机名)。可以把它想象成和HTTP头一样提供相同的功能---通过该方法传输正文之外的额外信息。Flume提供不同的source会给其他生成的event添加不同的header。

5 Agent(代理)

(1) Flume内部有一个或多个Agent；

(2) 每一个Agent是一个独立的守护进程(JVM)；

(3) 从客户端收集数据，或者从其他Agent那接收数据，然后迅速的将获得的数据传给下一个节点的Agent；

(4) Agent主要由source、channel、sink三个组件组成，必须要有的，可选的有interceptor、selector，根据业务配置。五者位置如下。

6 Source(数据源)

(1) Source是一个Flume源；

(2) 负责一个外部源(数据发生器)数据接收，如一个web服务器传递给它的事件；

(3) Source将接收的数据转换成flume可以识别的格式，这个格式就是event；

(4) 当一个Flume源接收到一个事件时，其将通过一个或者多个通道存储该事件。

7 Channel(通道/存储)

(1) 事件传输的通道；

(2) 采用被动存储的形式，即通道会缓存该事件直到该事件被sink组件异步处理；

(3) Channel是一种短暂的存储容器，它将从source处接收到的event格式的数据存储起来，直到它们被sinks消费掉，它在source和sink间起着一共桥梁的作用，channel是一个完整的事务，这一点保证了数据在收发的时候的一致性，并且它可以和任意数量的source和sink链接；

(4) 可以通过参数设置event的最大个数，流量控制。

(5) 数据从source接收后，会先在channel做缓存，缓存形式有两种，文件形式(FileChannel)和内存形式(Memeory Channel)：

Memory Channel:内存存储事务，吞吐率极高，，速度快，但存在丢数据风险；
Flie Channel:本地磁盘的事务实现模式，保证数据不会丢失(WAL实现)。

(6) Flume通常选择FileChannel，而不使用MemeoryChannel；

(7) 存储到内存还是存储到文件可通过配置文件去配置，建议用文件形式；

(8) 当数据传输完成后，该事件才能被通道移除，保证可靠性。

8 Sink(槽)

(1) Sink会将事件从Channel中移除，并将event放置到外部数据介质(HDFS)上

(2) 例如：通过Flume HDFS Sink将数据放置到HDFS中，或者放置到下一个Flume的Source，等到下一个Flume处理。

(3) 对于缓存在通道中的事件，Source和Sink采用异步处理的方式

(4) Sink成功取出Event后，将Event从channel中移除

(5) Sink必须作用于一个确切的Channel

(6) 数据以哪种存储形式落地是由不同类型的Sink决定的：

存储event到最终目的的终端:HDFS、Hbase
自动消耗:Null Sink
用于Agent之间通信：Avro

9 Interceptor(拦截器)

(1) interceptor是在你的数据源和你的source之间的一个环节，那这个环节相当于就是可以对你的数据源提前会做一个过滤；

(2) Interceptor用于Source的一组拦截器，按照预设的顺序必要地方对events进行过滤和自定义的处理逻辑实现；

(3) 在app(应用程序日志)和source之间的，对app日志进行拦截处理的。也即在日志进入source之前，对日志进行一些包装、清新过滤等等动作；

(4) 官方提供的已有的拦截器有：

Timestamp Interceptor：在event的header中添加一个key叫：timestamp,value为当前的时间戳；

Host Interceptor：在event的header中添加一个key叫：host,value为当前机器的hostname或者ip；

Static Interceptor：可以在event的header中添加自定义的key和value；

Regex Filtering Interceptor：通过正则来清洗或包含匹配的events；

Regex Extractor Interceptor：通过正则表达式来在header中添加指定的key,value则为正则匹配的部分；

(5) Flume的拦截器也是chain(链)形式的，即可以对一个source指定多个拦截器，按先后顺序依次处理。

(6) 这里的拦截并不等于过滤。

10 selector(通道选择器)

(1) selector是有点类似于路由选择

(2) 位于source和channel之间，故称通道选择器，主要有两种类型:

Replicating(复制) ChannelSelector (default)：将source过来的events发往所有channel；

Multiplexing(复用) ChannelSelector：而Multiplexing 可以选择该发往哪些channel。

(3) 对于有选择性选择数据源，明显需要使用Multiplexing这种分发方式

(4) 问题：Multiplexing需要判断header里指定key的值来决定分发到某个具体的channel，如果demo和demo2同时运行在同一个服务器上，如果在不同的服务器上运行，可以在source1上加上一个host拦截器，这样可以通过header中的host来判断event该分发给哪个channel，而这里是在同一个服务器上，由于host是是区分不出日志来源的，必须想办法在header中添加一个key来区分日志的来源。

可以通过设置上游不同的Source解决问题。

(5) 广播：复制出多份信息，下游每一个节点都同时接收同样的消息。