Flume---大数据协做框架

时间 2019-12-10

原文原文链接

flume是什么

Apache Flume是一个分布式的、可靠的、易用的系统，能够有效地未来自不少不一样源系统的大量日志数据收集、汇总或者转移到一个数据中心存储。html

Apache Flume的做用不只限于日志汇总，由于数据源是能够自定义的，Flume也能够被用于传输大量的事件数据，包括但不限于网络流量数据、社交媒体产生的数据、电子邮件和几乎全部可能的数据源。node

安装flume

tar -zxvf flume-ng-1.5.0-cdh5.3.6.tar.gz -C /opt/cdh-5.3.6
vi flume-env.sh
export JAVA_HOME=/opt/modules/jdk1.7.0_67

架构图

flume特色

flume的数据流由事件(Event)贯穿始终。事件是Flume的基本数据单位，它携带日志数据(字节数组形式)而且携带有头信息，这些Event由Agent外部的Source生成，当Source捕获事件后会进行特定的格式化，而后Source会把事件推入(单个或多个)Channel中。能够把Channel看做是一个缓冲区，它将保存事件直到Sink处理完该事件。Sink负责持久化日志或者把事件推向另外一个Source。web

flume可靠性

当节点出现故障时，日志可以被传送到其余节点上而不会丢失。Flume提供了三种级别的可靠性保障，从强到弱依次分别为：end-to-end（收到数据agent首先将event写到磁盘上，当数据传送成功后，再删除；若是数据发送失败，能够从新发送。），Store on failure（这也是scribe采用的策略，当数据接收方crash时，将数据写到本地，待恢复后，继续发送），Besteffort（数据发送到接收方后，不会进行确认）。shell

flume的一些核心概念

Agent使用JVM 运行Flume。每台机器运行一个agent，可是能够在一个agent中包含多个sources和sinks。数据库

Client生产数据，运行在一个独立的线程。apache

Source从Client收集数据，传递给Channel。bootstrap

Sink从Channel收集数据，运行在一个独立线程。数组

Channel链接 sources 和 sinks ，这个有点像一个队列。安全

Events能够是日志记录、 avro 对象等。bash

Flume提供了大量内置的Source、Channel和Sink类型。不一样类型的Source,Channel和Sink能够自由组合。组合方式基于用户设置的配置文件，很是灵活。好比：Channel能够把事件暂存在内存里，也能够持久化到本地硬盘上。Sink能够把日志写入HDFS, HBase，甚至是另一个Source等等

flume的使用

设置代理
Flume代理配置存储在本地配置文件中。这是一个遵循Java属性文件格式的文本文件。一个或多个代理的配置能够在同一个配置文件中指定。配置文件包括代理中每一个源，宿和通道的属性以及它们如何链接在一块儿以造成数据流。
小例子

＃example.conf：单节点Flume配置

＃将该代理商的组件命名为
a1.sources  =  r1 
a1.sinks  =  k1 
a1.channels  =  c1

＃描述/配置源
a1.sources.r1.type  =  netcat 
a1.sources.r1.bind  =  hadoop.jianxin.com 
a1.sources.r1.port  =  44444

＃描述sink 
a1.sinks.k1.type  =  logger

＃使用缓冲内存中事件的通道
a1.channels.c1.type  =  memory 
a1.channels.c1.capacity  =  1000 
a1.channels.c1.transactionCapacity  =  100

＃将信源和信宿绑定到信道
a1.sources.r1.channels  =  c1 
a1.sinks.k1.channel  =  c1

启动方式

bin / flume-ng agent --conf conf --conf -file example.conf --name a1 -Dflume.root.logger = INFO，console

经常使用的source

NetCat source:用来监听一个指定端口，并将接收到的数据的每一行转换为一个事件。
avro source:avro能够监听和收集指定端口的日志，使用avro的source须要说明被监听的主机ip和端口号

＃将该代理商的组件命名为
a1.sources  =  r1 
a1.sinks  =  k1 
a1.channels  =  c1

＃描述/配置源
a1.sources.r1.type  =  avro 
a1.sources.r1.bind  =  hadoop.jianxin.com 
a1.sources.r1.port  =  3306

＃描述sink 
a1.sinks.k1.type  =  logger

＃使用缓冲内存中事件的通道
a1.channels.c1.type  =  memory 
a1.channels.c1.capacity  =  100 
a1.channels.c1.transactionCapacity  =  10

＃将信源和信宿绑定到信道
a1.sources.r1.channels  =  c1 
a1.sinks.k1.channel  =  c1

exec source:能够经过指定的操做对日志进行读取，使用exec时须要指定shell命令，对日志进行读取
Spooling-directory source:能够读取文件夹里的日志，使用时指定一个文件夹，能够读取该文件夹中的全部文件，当出现新文件时会读取该文件并获取数据.须要注意的是该文件夹中的文件在读取过程当中不能修改，同时文件名也不能修改。

spoolDirectory是监控目录，不能为空，没有默认值。这个source不具备监控子目录的功能，也就是不能递归监控。若是须要，这须要本身去实现，http://blog.csdn.net/yangbutao/article/details/8835563 这里有递归检测的实现；
completedSuffix是文件读取完毕后给完成文件添加的标记后缀，默认是".COMPLETED"；
deletePolicy这是是否删除读取完毕的文件，默认是"never"，就是不删除，目前只支持"never"和“IMMEDIATE”；
fileHeader是否在event的Header中添加文件名，boolean类型, 默认false
fileHeaderKey这是event的Header中的key,value是文件名
batchSize这个是一次处理的记录数，默认是100；
inputCharset编码方式，默认是"UTF-8"；
ignorePattern忽略符合条件的文件名
trackerDirPath被处理文件元数据的存储目录，默认".flumespool"
deserializerType将文件中的数据序列化成event的方式，默认是“LINE”---org.apache.flume.serialization.LineDeserializer
deserializerContext这个主要用在Deserializer中设置编码方式outputCharset和文件每行最大长度maxLineLength。

常见的channel

memory channel:内存

a1.channels = c1
a1.channels.c1.type = memory
a1.channels.c1.capacity = 10000
a1.channels.c1.transactionCapacity = 10000
a1.channels.c1.byteCapacityBufferPercentage = 20
a1.channels.c1.byteCapacity = 800000

Kafka Channel

a1.channels.channel1.type = org.apache.flume.channel.kafka.KafkaChannel
a1.channels.channel1.kafka.bootstrap.servers = kafka-1:9092,kafka-2:9092,kafka-3:9092
a1.channels.channel1.kafka.topic = channel1
a1.channels.channel1.kafka.consumer.group.id = flume-consumer

File Channel

常见的sink

logger sink:将收集到的日志写到flume的log
avro sink:能够将接受到的日志发送到指定端口，供级联agent的下一跳收集和接受日志，使用时须要指定目的ip和端口

a1.sinks=k1

a1.sinks.k2.type = avro

a1.sinks.k2.channel = c2

a1.sinks.k2.hostname = hadoop03 (指定的主机名或ip)

a1.sinks.k2.port = 16666  (指定的端口号)

hdfs sink:将收集到的日志写入到新建立的文件中保存起来，存储路径为分布式的文件系统hdfs的路径，同时hdfs建立新文件的周期能够是时间，也能够是文件的大小，还能够是采集日志的条数

path:写入hdfs的路径，须要包含文件系统标识，好比：hdfs://namenode/flume/webdata/

可使用flume提供的日期及%{host}表达式。

filePrefix： 默认值：FlumeData 写入hdfs的文件名前缀，可使用flume提供的日期及%{host}表达式。

fileSuffix：写入hdfs的文件名后缀，好比：.lzo .log等。

inUsePrefix：临时文件的文件名前缀，hdfs sink会先往目标目录中写临时文件，再根据相关规则重命名成最终目标文件；

inUseSuffi: 默认值：.tmp  临时文件的文件名后缀。

rollInterval:  默认值：30  hdfs sink间隔多长将临时文件滚动成最终目标文件，单位：秒；

若是设置成0，则表示不根据时间来滚动文件；

注：滚动（roll）指的是，hdfs sink将临时文件重命名成最终目标文件，并新打开一个临时文件来写入数据；

rollSize  默认值：1024 当临时文件达到该大小（单位：bytes）时，滚动成目标文件；

若是设置成0，则表示不根据临时文件大小来滚动文件；

rollCount  默认值：10  当events数据达到该数量时候，将临时文件滚动成目标文件；

若是设置成0，则表示不根据events数据来滚动文件；

idleTimeout  默认值：0

当目前被打开的临时文件在该参数指定的时间（秒）内，没有任何数据写入，则将该临时文件关闭并重命名成目标文件；

batchSize   默认值：100  每一个批次刷新到HDFS上的events数量；

codeC  文件压缩格式，包括：gzip, bzip2, lzo, lzop, snappy

fileType  默认值：SequenceFile

文件格式，包括：SequenceFile, DataStream,CompressedStream 当使用DataStream时候，文件不会被压缩，不须要设置hdfs.codeC;当使用CompressedStream时候，必须设置一个正确的hdfs.codeC值；

maxOpenFiles  默认值：5000 最大容许打开的HDFS文件数，当打开的文件数达到该值，最先打开的文件将会被关闭；

minBlockReplicas  默认值：HDFS副本数   写入HDFS文件块的最小副本数。该参数会影响文件的滚动配置，通常将该参数配置成1，才能够按照配置正确滚动文件。

writeFormat  写sequence文件的格式。包含：Text, Writable（默认）

callTimeout  默认值：10000 执行HDFS操做的超时时间（单位：毫秒）；

threadsPoolSize  默认值：10  hdfs sink启动的操做HDFS的线程数。

rollTimerPoolSize  默认值：1  hdfs sink启动的根据时间滚动文件的线程数。

kerberosPrincipal： HDFS安全认证kerberos配置；

kerberosKeytab

HDFS安全认证kerberos配置；

proxyUser  代理用户

round 默认值：false  是否启用时间上的”舍弃”，这里的”舍弃”，相似于”四舍五入”，后面再介绍。若是启用，则会影响除了%t的其余全部时间表达式；

roundValue  默认值：1   时间上进行“舍弃”的值；

roundUnit  默认值：seconds 时间上进行”舍弃”的单位，包含：second,minute,hour


a1.sinks.k1.hdfs.path = /flume/events/%y-%m-%d/%H%M/%S

a1.sinks.k1.hdfs.round = true

a1.sinks.k1.hdfs.roundValue = 10

a1.sinks.k1.hdfs.roundUnit = minute

当时间为2015-10-16 17:38:59时候，hdfs.path依然会被解析为：

/flume/events/20151016/17:30/00

由于设置的是舍弃10分钟内的时间，所以，该目录每10分钟新生成一个。

timeZone

默认值：Local Time  时区。

useLocalTimeStamp 默认值：flase  是否使用当地时间。

closeTries：默认值：0

hdfs sink关闭文件的尝试次数；

若是设置为1，当一次关闭文件失败后，hdfs sink将不会再次尝试关闭文件，这个未关闭的文件将会一直留在那，而且是打开状态。

设置为0，当一次关闭失败后，hdfs sink会继续尝试下一次关闭，直到成功。

retryInterval

默认值：180（秒）

hdfs sink尝试关闭文件的时间间隔，若是设置为0，表示不尝试，至关于于将hdfs.closeTries设置成1.

Serializer  默认值：TEXT

序列化类型。其余还有：avro_event或者是实现了EventSerializer.Builder的类名。

Hbase sink:hbase是一种数据库，能够储存日志，使用时须要指定存储日志的表名和列族名，而后agent就能够将收集到的日志逐条插入到数据库中。

tableName：要写入的HBase数据表名，不能为空；

columnFamily：数据表对应的列簇名，这个sink目前只支持一个列簇，不能为空；

batchSize：每次事务能够处理的最大Event数量，默认是100；

eventSerializerType：用来将event写入HBase，即将event转化为put。默认是org.apache.flume.sink.hbase.SimpleHbaseEventSerializer，还有一个是RegexHbaseEventSerializer，即适合HBaseSink的Serializer只有这俩，不然本身定制；

serializerContext：是eventSerializerType的配置信息，就是配置文件中包含“serializer.”的项；

例子：

agent1.sinks = k1

agent1.sinks.k1.type = hbase

agent1.sinks.k1.table = flume

agent1.sinks.k1.columnFamily=fl_conf

agent1.sinks.k1.serializer=org.apache.flume.sink.hbase.RegexHbaseEventSerializer

综合小案例

＃将该代理商的组件命名为
a1.sources  =  r1 
a1.sinks  =  k1 
a1.channels  =  c1

＃描述/配置源
a1.sources.r1.type  =  avro 
a1.sources.r1.bind  =  hadoop.jianxin.com 
a1.sources.r1.port  =  3306

＃描述sink 
a1.sinks.k1.type  =  logger

＃使用缓冲内存中事件的通道
a1.channels.c1.type  =  memory 
a1.channels.c1.capacity  =  100 
a1.channels.c1.transactionCapacity  =  10

＃将信源和信宿绑定到信道
a1.sources.r1.channels  =  c1 
a1.sinks.k1.channel  =  c1



a2.sources=r2
a2.sinks=k2
a2.channels=c2

#描述/配置源
a2.sources.r2.type=exec
a2.sources.r2.command=tail -f /opt/cdh-5.3.6/hive-0.13.1-cdh5.3.6/logs/hive.log
a2.sources.r2.shell=/bin/bash -c

＃使用缓冲内存中事件的通道
a2.channels.c2.type=memory
a1.channels.c2.capacity  =  100 
a1.channels.c2.transactionCapacity  =  10

＃描述sink 
a2.sinks.k2.type=hdfs
a2.sinks.k2.hdfs.path=hdfs://hadoop.jianxin.com:9000/user/jianxin/flume/hive/date/%Y%m%d
a2.sinks.k2.hdfs.fileType=DataStream 
a2.sinks.k2.hdfs.writeFormat=Text 
#Format for sequence file records. One of Text or Writable. 
#Set to Text before creating data files with Flume, 
#otherwise those files cannot be read by either Apache Impala (incubating) or Apache Hive.
a2.sinks.k2.hdfs.batchSize=10
#number of events written to file before it is flushed to HDFS
a2.sinks.k2.hdfs.useLocalTimeStamp=true
＃将信源和信宿绑定到信道
a2.sources.r2.channels=c2
a2.sinks.k2.channel=c2

运行小案例

#启动的会是a2source
bin/flume-ng agent --conf conf --conf-file conf/conf.file --name a2 -Dflume.root.logger=INFO,console
nohup bin/flume-ng agent --conf conf --conf-file conf/conf.file --name a2 -Dflume.root.logger=INFO,console > /data/flume-1.5.0-cdh5.3.6/flume.log 2>&1 &

参考 https://blog.csdn.net/wei_hhh/article/details/77838999 flume 官网 http://flume.apache.org/FlumeUserGuide.html