五、Flume Channel
Channel主要是用来缓冲Agent以及接受，但尚未写出到另外一个Agent或者存储系统的数据。Channel的行为比较像队列，Source写入到他们，Sink从他们中读取数据。多个Source可以安全的写入到同一Channel中，并且多个Sink可以从同一个Channel中读取数据。可是一个Sink只能从一个Channel读取数据，如果多个Sink从相同的Channel中读取数据，系统可以保证只有一个Sink会从Channel读取一个特定的事件。
关于channel的配置见 官网channel配置介绍
常见Flume Channel的分类
Channel类型 说明
Memory Channel Event数据存储在内存中
JDBC Channel Event数据存储在持久化存储中，当前Flume Channel内置支持Derby
File Channel Event数据存储在磁盘文件中
Spillable Memory Channel Event数据存储在内存中和磁盘上，当内存队列满了，会持久化到磁盘文件
Pseudo Transaction Channel 测试用途
Custom Channel 自定义Channel实现
六、Flume Sinks
Sink会连续轮训各自的Channel来读取和删除事件。Sink将事件推送到下一阶段（RPC Sink的情况下），或者到达最终目的地。一旦在下一阶段或者其目的地中数据是安全的,Sink通过事务提交通知Channel，可以从Channel中删除这一事件。
所有sink类型如下 ,下面介绍一些主要的sink
HDFS Sink
这个接收器将事件写入Hadoop分布式文件系统(HDFS)。它目前支持创建文本和序列文件。它支持两种文件类型的压缩。可以根据经过的时间、数据大小或事件数周期性地滚动文件(关闭当前文件并创建新文件)。它还根据事件起源的时间戳或机器等属性对数据进行存储/分区。HDFS目录路径可能包含格式转义序列，这些转义序列将被HDFS接收器替换，以生成目录/文件名来存储事件。使用此接收器需要安装Hadoop，以便Flume可以使用HadoopJAR与HDFS集群通信。注意，支持sync()调用的Hadoop版本是必需的。
配置参数
注意
正在使用的文件的名称将经过修饰，以末尾包含“ .tmp”。关闭文件后，将删除此扩展名。这样可以排除目录中的部分完整文件。必需的属性以粗体显示。
对于所有与时间相关的转义序列，事件的标头中必须存在带有键“ timestamp”的标头（除非hdfs.useLocalTimeStamp设置为true）。一种自动添加此方法的方法是使用TimestampInterceptor。
配置参数
案例演示
创建flume 的自定义配置文件 hdfs-sink
a1.sources = r1 a1.sinks = k1 a1.channels = c1 # Describe/configure the source a1.sources.r1.type = spooldir a1.sources.r1.spoolDir = /home/logs a1.sources.r1.fileHeader = true # Describe the sink a1.sinks.k1.type = logger a1.sinks.k1.type=hdfs # 这里注意flume注册时会自动加载hdfs, 因此可以不指定hdfs的路径 a1.sinks.k1.hdfs.path=/flume/events/%Y-%m-%d/%H%M ##每隔60s或者文件大小超过10M的时候产生新文件 # hdfs有多少条消息时新建文件，0不基于消息个数 a1.sinks.k1.hdfs.rollCount=0 # hdfs创建多长时间新建文件，0不基于时间 a1.sinks.k1.hdfs.rollInterval=60 # hdfs多大时新建文件，0不基于文件大小 a1.sinks.k1.hdfs.rollSize=10240 # 当目前被打开的临时文件在该参数指定的时间（秒）内，没有任何数据写入，则将该临时文件关闭并重命名成目标文件 a1.sinks.k1.hdfs.idleTimeout=3 ## 每五分钟生成一个目录: # 是否启用时间上的”舍弃”，这里的”舍弃”，类似于”四舍五入”，后面再介绍。如果启用，则会影响除了%t的其他所有时间表达式 a1.sinks.k1.hdfs.round=true # 时间上进行“舍弃”的值； a1.sinks.k1.hdfs.roundValue=5 # 时间上进行”舍弃”的单位，包含：second,minute,hour a1.sinks.k1.hdfs.roundUnit=second a1.sinks.k1.hdfs.fileType=DataStream a1.sinks.k1.hdfs.useLocalTimeStamp=true # 设置超时时间 a1.sinks.k1.hdfs.callTimeout=60000 # 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
根据自定义文件a1.sources.r1.spoolDir = /home/logs,创建目录 ./home/logs
运行 flume flume-ng agent --conf-file hdfs-sink --name a1 -Dflume.root.logger=INFO,console
移动任意日志文件到 /home/logs 目录下, 效果如图1, 图2所示
图1
flume 阻塞式界面输出相关信息
图2
可以看到日志被以时间的顺序读取到hdfs目录下
注意: 关于其他sink的配置见官网 官网介绍如下
flume在项目中的应用
flume读取指定目录文件(nginx的指定日志文件 这里是Nginx的配置 )下的数据
并将其收集保存在本地具体实现步骤如下
编写 flume的自定义配置文件 ,文件名 project
这里指定了读取nginx 的访问日志文件/opt/data/access.log
以及读取后的文件在hdfs的中的目录/log/%Y%m%d ,%Y%m%d是文件前面的目录名为当前日期
idleTimeout = 10 代表10s内如果没有文件传输, 自动关闭文件该文件的写入功能 ,10s再写入会被写入到另一个文件中
a1.sources = r1 a1.sinks = k1 a1.channels = c1 # Describe/configure the source a1.sources.r1.type = spooldir a1.sources.r1.spoolDir = /home/logs a1.sources.r1.fileHeader = true # Describe the sink ## 指定sink为hdfs, 即从hdfs那里接受channel中的数据, 并指定hdfs的相关目录 a1.sinks.k1.type=hdfs a1.sinks.k1.hdfs.path=hdfs://logs/flume/%Y-%m-%d/%H%M ##每隔60s或者文件大小超过10M的时候产生新文件 # hdfs有多少条消息时新建文件，0不基于消息个数 a1.sinks.k1.hdfs.rollCount=0 # hdfs创建多长时间新建文件，0不基于时间,时间单位 s a1.sinks.k1.hdfs.rollInterval=60 # hdfs多大时新建文件，0不基于文件大小 a1.sinks.k1.hdfs.rollSize=10240 # 当目前被打开的临时文件在该参数指定的时间（秒）内，没有任何数据写入，则将该临时文件关闭并重命名成目标文件 a1.sinks.k1.hdfs.idleTimeout=3 a1.sinks.k1.hdfs.fileType=DataStream a1.sinks.k1.hdfs.useLocalTimeStamp=true ## 每五分钟生成一个目录: # 是否启用时间上的”舍弃”，这里的”舍弃”，类似于”四舍五入”，后面再介绍。如果启用，则会影响除了%t的其他所有时间表达式 a1.sinks.k1.hdfs.round=true # 时间上进行“舍弃”的值； a1.sinks.k1.hdfs.roundValue=5 # 时间上进行”舍弃”的单位，包含：second,minute,hour a1.sinks.k1.hdfs.roundUnit=minute # 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
通过该自定义配置文件运行 flume
flume-ng agent --conf-file project --name a1 -Dflume.root.logger=INFO,console
