标签:direct format 之间 netca channels interval 创建目录 级别 .bat
Flume是Cloudera提供的一个高可用的,高可靠的,分布式的海量日志采集、聚合和传输的系统,目前是Apache的顶级项目。Flume支持在日志系统中定制各类数据发送方,用于收集数据;同时,Flume提供对数据进行简单处理,并写到各种数据接受方(可定制)的能力。
1、可靠性
当节点出现故障时,日志能够被传送到其他节点上而不会丢失。Flume提供了三种级别的可靠性保障,从强到弱依次分别为:
2、可扩展性
Flume采用了三层架构,分别为agent,collector和storage,每一层均可以水平扩展。
其中,所有agent和collector由master统一管理,这使得系统容易监控和维护,且master允许有多个(使用ZooKeeper进行管理和负载均衡),这就避免了单点故障问题。
3、可管理性
4、功能可扩展性
5、文档丰富,社区活跃
Flume 已经成为 Hadoop 生态系统的标配,它的文档比较丰富,社区比较活跃,方便我们学习。
Flume agent每个agent是一个独立的Java进程,从客户端(其他agent)接收数据然后转发到下一个destination(sink(沉槽) | agent)
Agent包含三个组件:
从事件生成器接收数据,以event事件的形式传给一个或多个channel
从source中接受flume event,作为临时存放地,缓存到buffer中,直到sink
将其消费掉,是source和sink之间的桥梁
Channel是事务的,可以和多个source或sink协同
存放数据到HDFS,从channel中消费event,并分发给destination,sink的
Destination 也可以是另一个agent或者HDFS,HBASE
注意:一个flume的agent,可以有多个source,channel,sink
Source: 完成对日志数据的收集,分成transtion 和 event 打入到channel之中, Flume提供了各种source的实现,包括Avro Source、 Exce Source、 Spooling
Directory Source、 NetCat Source、 Syslog Source、 Syslog TCP Source、Syslog UDP Source、 HTTP Source、 HDFS Source, etc。
Channel: Channel用于连接Source和Sink,Source将日志信息发送到Channel,Sink从Channel消费日志信息;Channel是中转日志信息的一个临时存储,保存有Source组件传递过来的日志信息。, flume提供了Memory Channel、 JDBC Chanel、 File Channel,etc
Sink: Flume Sink取出Channel中的数据,进行相应的存储文件系统,数据库,或者提交到远程服务器。包括HDFS sink、 Logger sink、 Avro sink、 File Roll sink、 Null sink、 HBasesink, etc。
# 下载
get http://mirrors.tuna.tsinghua.edu.cn/apache/flume/1.8.0/apache-flume-1.8.0-bin.tar.gz
# 解压
tar –zxvf apache-flume-1.8.0-bin.tar.gz
# 添加配置文件(读取指定文件写入HDFS中)
# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /export/logs/web.log
# Describe the sink
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path =hdfs://node-1:9000/app_log/%y-%m-%d/%H-%M
# 保存到HDFS上的前缀
a1.sinks.k1.hdfs.filePrefix = weichat_log
a1.sinks.k1.hdfs.fileSuffix = .dat
a1.sinks.k1.hdfs.batchSize= 100
a1.sinks.k1.hdfs.fileType = DataStream
a1.sinks.k1.hdfs.writeFormat =Text
# 配置存储在HDFS上的文件大小单位(bytes)
a1.sinks.k1.hdfs.rollSize = 262144
# 写入多少个event数据后滚动文件(事件个数)
a1.sinks.k1.hdfs.rollCount = 10
# 文件滚动之前的等待时间(秒)
a1.sinks.k1.hdfs.rollInterval = 120
# 1分钟就改目录(创建目录)
a1.sinks.k1.hdfs.round = true
a1.sinks.k1.hdfs.roundValue = 1
a1.sinks.k1.hdfs.roundUnit = minute
a1.sinks.k1.hdfs.useLocalTimeStamp = true
# 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
# 添加配置文件(读取指定目录写入HDFS中)
#定义三大组件的名称
ag1.sources = source1
ag1.sinks = sink1
ag1.channels = channel1
# 配置source组件
ag1.sources.source1.type = spooldir
ag1.sources.source1.spoolDir = /export/logs/
ag1.sources.source1.fileSuffix=.FINISHED
ag1.sources.source1.deserializer.maxLineLength=5120
# 配置sink组件
ag1.sinks.sink1.type = hdfs
ag1.sinks.sink1.hdfs.path =hdfs://hdp-01:9000/access_log/%y-%m-%d/%H-%M
ag1.sinks.sink1.hdfs.filePrefix = app_log
ag1.sinks.sink1.hdfs.fileSuffix = .log
ag1.sinks.sink1.hdfs.batchSize= 100
ag1.sinks.sink1.hdfs.fileType = DataStream
ag1.sinks.sink1.hdfs.writeFormat =Text
## roll:滚动切换:控制写文件的切换规则
## 按文件体积(字节)来切
ag1.sinks.sink1.hdfs.rollSize = 512000
## 按event条数切
ag1.sinks.sink1.hdfs.rollCount = 1000000
## 按时间间隔切换文件
ag1.sinks.sink1.hdfs.rollInterval = 60
## 控制生成目录的规则
ag1.sinks.sink1.hdfs.round = true
ag1.sinks.sink1.hdfs.roundValue = 10
ag1.sinks.sink1.hdfs.roundUnit = minute
ag1.sinks.sink1.hdfs.useLocalTimeStamp = true
# channel组件配置
ag1.channels.channel1.type = memory
## event条数
ag1.channels.channel1.capacity = 500000
##flume事务控制所需要的缓存容量600条event
ag1.channels.channel1.transactionCapacity = 600
# 绑定source、channel和sink之间的连接
ag1.sources.source1.channels = channel1
ag1.sinks.sink1.channel = channel1
# 启动flume
$FLUME_HOME/bin/flume-ng agent -c conf -f conf/tail-hdfs.conf -n a1 -Dflume.root.logger=INFO,console
标签:direct format 之间 netca channels interval 创建目录 级别 .bat
原文地址:https://www.cnblogs.com/ruolin/p/10209767.html