官方参考文档
Flume NG是一个分布式、可靠、可用的系统，它能够将不同数据源的海量日志数据进行高效收集、聚合、移动，最后存储到一个中心化数据存储系统中。由原来的Flume OG到现在的Flume NG，进行了架构重构，并且现在NG版本完全不兼容原来的OG版本。经过架构重构后，Flume NG更像是一个轻量的小工具，非常简单，容易适应各种方式日志收集，并支持failover和负载均衡。
架构设计要点
Flume的架构主要有一下几个核心概念：
Event：一个数据单元，带有一个可选的消息头
Flow：Event从源点到达目的点的迁移的抽象
Client：操作位于源点处的Event，将其发送到Flume Agent
Agent：一个独立的Flume进程，包含组件Source、Channel、Sink
Source：用来消费传递到该组件的Event
Channel：中转Event的一个临时存储，保存有Source组件传递过来的Event
Sink：从Channel中读取并移除Event，将Event传递到Flow Pipeline中的下一个Agent（如果有的话）
Flume NG架构，如图所示：
外部系统产生日志，直接通过Flume的Agent的Source组件将事件（如日志行）发送到中间临时的channel组件，最后传递给Sink组件，HDFS Sink组件可以直接把数据存储到HDFS集群上。
一个最基本Flow的配置，格式如下：
# list the sources, sinks and channels for the agent.sources = .sinks = .channels = # set channel for source .sources. .channels = ... .sources. .channels = ...# set channel for sink .sinks. .channel = .sinks. .channel =
尖括号里面的，我们可以根据实际需求或业务来修改名称。下面详细说明：
表示配置一个Agent的名称，一个Agent肯定有一个名称。与是Agent的Source组件的名称，消费传递过来的Event。与是Agent的Channel组件的名称。与是Agent的Sink组件的名称，从Channel中消费（移除）Event。上面配置内容中，第一组中配置Source、Sink、Channel，它们的值可以有1个或者多个；第二组中配置Source将把数据存储（Put）到哪一个Channel中，可以存储到1个或多个Channel中，同一个Source将数据存储到多个Channel中，实际上是Replication；第三组中配置Sink从哪一个Channel中取（Task）数据，一个Sink只能从一个Channel中取数据。下面，根据官网文档，我们展示几种Flow Pipeline，各自适应于什么样的应用场景：多个Agent顺序连接
多个Agent的数据汇聚到同一个Agent
多路（Multiplexing）Agent
# List the sources, sinks and channels for the agent.sources = .sinks = .channels = # set list of channels for source (separated by space) .sources. .channels = # set channel for sinks .sinks. .channel = .sinks. .channel = .sources. .selector.type = replicating
上面指定了selector的type的值为replication，其他的配置没有指定，使用的Replication方式，Source1会将数据分别存储到Channel1和Channel2，这两个channel里面存储的数据是相同的，然后数据被传递到Sink1和Sink2。
Multiplexing方式，selector可以根据header的值来确定数据传递到哪一个channel，配置格式，如下所示：
# Mapping for multiplexing selector.sources. .selector.type = multiplexing .sources. .selector.header = .sources. .selector.mapping. = .sources. .selector.mapping. = .sources. .selector.mapping. = #... .sources. .selector.default =
上面selector的type的值为multiplexing，同时配置selector的header信息，还配置了多个selector的mapping的值，即header的值：如果header的值为Value1、Value2，数据从Source1路由到Channel1；如果header的值为Value2、Value3，数据从Source1路由到Channel2。
实现load balance功能
a1.sinkgroups = g1a1.sinkgroups.g1.sinks = k1 k2 k3a1.sinkgroups.g1.processor.type = load_balancea1.sinkgroups.g1.processor.backoff = truea1.sinkgroups.g1.processor.selector = round_robina1.sinkgroups.g1.processor.selector.maxTimeOut=10000
实现failover能
Failover Sink Processor能够实现failover功能，具体流程类似load balance，但是内部处理机制与load balance完全不同：Failover Sink Processor维护一个优先级Sink组件列表，只要有一个Sink组件可用，Event就被传递到下一个组件。如果一个Sink能够成功处理Event，则会加入到一个Pool中，否则会被移出Pool并计算失败次数，设置一个惩罚因子，示例配置如下所示：
a1.sinkgroups = g1a1.sinkgroups.g1.sinks = k1 k2 k3a1.sinkgroups.g1.processor.type = failovera1.sinkgroups.g1.processor.priority.k1 = 5a1.sinkgroups.g1.processor.priority.k2 = 7a1.sinkgroups.g1.processor.priority.k3 = 6a1.sinkgroups.g1.processor.maxpenalty = 20000
基本功能
我们看一下，Flume NG都支持哪些功能（目前最新版本是1.5.0.1），了解它的功能集合，能够让我们在应用中更好地选择使用哪一种方案。说明Flume NG的功能，实际还是围绕着Agent的三个组件Source、Channel、Sink来看它能够支持哪些技术或协议。我们不再对各个组件支持的协议详细配置进行说明，通过列表的方式分别对三个组件进行概要说明：
Source类型 说明Avro Source 支持Avro协议（实际上是Avro RPC），内置支持Thrift Source 支持Thrift协议，内置支持Exec Source 基于Unix的command在标准输出上生产数据JMS Source 从JMS系统（消息、主题）中读取数据，ActiveMQ已经测试过Spooling Directory Source 监控指定目录内数据变更Twitter 1% firehose Source 通过API持续下载Twitter数据，试验性质Netcat Source 监控某个端口，将流经端口的每一个文本行数据作为Event输入Sequence Generator Source 序列生成器数据源，生产序列数据Syslog Sources 读取syslog数据，产生Event，支持UDP和TCP两种协议HTTP Source 基于HTTP POST或GET方式的数据源，支持JSON、BLOB表示形式Legacy Sources 兼容老的Flume OG中Source（0.9.x版本）Flume Channel###############################################################Channel类型 说明Memory Channel Event数据存储在内存中JDBC Channel Event数据存储在持久化存储中，当前Flume Channel内置支持DerbyFile Channel Event数据存储在磁盘文件中Spillable Memory Channel Event数据存储在内存中和磁盘上，当内存队列满了，会持久化到磁盘文件（当前试验性的，不建议生产环境使用）Pseudo Transaction Channel 测试用途Custom Channel 自定义Channel实现Flume Sink###################################################################Sink类型 说明HDFS Sink 数据写入HDFSLogger Sink 数据写入日志文件Avro Sink 数据被转换成Avro Event，然后发送到配置的RPC端口上Thrift Sink 数据被转换成Thrift Event，然后发送到配置的RPC端口上IRC Sink 数据在IRC上进行回放File Roll Sink 存储数据到本地文件系统Null Sink 丢弃到所有数据HBase Sink 数据写入HBase数据库Morphline Solr Sink 数据发送到Solr搜索服务器（集群）ElasticSearch Sink 数据发送到Elastic Search搜索服务器（集群）Kite Dataset Sink 写数据到Kite Dataset，试验性质的Custom Sink 自定义Sink实现#################################################################另外还有Channel Selector、Sink Processor、Event Serializer、Interceptor等组件，可以参考官网提供的用户手册。
安装配置略，可以参考网上教程
下面是测试的配置文件
agent 配置文件如下
# Name the components on this agenta1.sources = r1a1.sinks = k1a1.channels = c1a1.sinks.k1.type = avro a1.sinks.k1.hostname = 127.0.0.1a1.sinks.k1.port = 44444 a1.sinks.k1.channel = c1# Use a channel which buffers events in memorya1.channels.c1.type = memorya1.channels.c1.capacity = 1000a1.channels.c1.transactionCapacity = 100a1.sources.r1.type = execa1.sources.r1.command = tail -F /var/log/nginx/access.loga1.sources.r1.channels = c1a1.sources.s.deserializer.maxLineLength=65535
server端配置文件如下： 测试复制（Replication）1个source 复制到多个channels 输出到多个sink
# Name the components on this agentb1.sources = r1b1.sinks = k1 k2 k3 b1.channels = c1 c2 c3b1.sources.r1.selector.type = replicatingb1.sources.r1.type = avro b1.sources.r1.channels = c1 c2 c3 b1.sources.r1.bind = 0.0.0.0 b1.sources.r1.port = 44444b1.channels.c1.type = file b1.channels.c1.write-timeout = 10 b1.channels.c1.keep-alive = 10 b1.channels.c1.checkpointDir = /flume/checkb1.channels.c1.useDualCheckpoints = true b1.channels.c1.backupCheckpointDir = /flume/backupb1.channels.c1.dataDirs = /flumeb1.channels.c2.type=memory b1.channels.c2.capacity=2000000 b1.channels.c2.transactionCapacity=10000 b1.channels.c3.type=memory b1.channels.c3.capacity=2000000 b1.channels.c3.transactionCapacity=10000 # Describe the sink b1.sinks.k1.type = hdfsb1.sinks.k1.channel = c1b1.sinks.k1.hdfs.path = hdfs://localhost:9000/user/hadoop/flume/collected/b1.sinks.k1.hdfs.filePrefix = chen_testb1.sinks.k1.hdfs.round = trueb1.sinks.k1.hdfs.roundValue = 10b1.sinks.k1.hdfs.roundUnit = minuteb1.sinks.k2.channel = c2b1.sinks.k2.type = file_rollb1.sinks.k2.batchSize = 100000000b1.sinks.k2.rollInterval = 1000000b1.sinks.k2.serializer = TEXTb1.sinks.k2.sink.directory = /var/log/flumeb1.sinks.k3.channel = c3 b1.sinks.k3.type = logger
启动测试命令
flume-ng agent -c . -f test.conf -n b1 -Dflume.root.logger=INFO,console
-c 配置文件目录 -f配置文件 -n 节点名字 和配置文件对应 console打到终端
5.
参考链接
6. 测试 区分 flume日志合并在一起的日志
a1配置[root@host_12 test]# cat a1.conf # Name the components on this agenta1.sources = r1a1.sinks = k1a1.channels = c1a1.sources.r1.interceptors = i1a1.sources.r1.interceptors.i1.type=statica1.sources.r1.interceptors.i1.key=nginxa1.sources.r1.interceptors.i1.value=nginx_1a1.sources.r1.interceptors.i1.preserveExisting=false#a1.sources.r1.interceptors = i1#a1.sources.r1.interceptors.i1.type = host#a1.sources.r1.interceptors.i1.hostHeader = hostnamea1.sinks.k1.type = avro a1.sinks.k1.hostname = 127.0.0.1a1.sinks.k1.port = 44444 a1.sinks.k1.channel = c1# Use a channel which buffers events in memorya1.channels.c1.type = memorya1.channels.c1.capacity = 1000a1.channels.c1.transactionCapacity = 100a1.sources.r1.type = exec###匹配shell tomcat yyyy:mm:dd:hh格式的日志a1.sources.r1.shell = /bin/bash -ca1.sources.r1.command = tail -f /var/log/nginx_1/access_`date +%Y%m%d%H`.loga1.sources.r1.channels = c1#########匹配替换行里的文本的内容#a1.sources.r1.interceptors = i1#a1.sources.r1.interceptors.i1.type = search_replace#a1.sources.r1.interceptors.i1.searchPattern = [0-9]+#a1.sources.r1.interceptors.i1.replaceString = lxw1234#a1.sources.r1.interceptors.i1.charset = UTF-8###########################################a2配置[root@host_12 test]# cat a2.conf # Name the components on this agenta2.sources = r1a2.sinks = k1a2.channels = c1a2.sources.r1.interceptors = i1a2.sources.r1.interceptors.i1.type=statica2.sources.r1.interceptors.i1.key=nginxa2.sources.r1.interceptors.i1.value=nginx_2a2.sources.r1.interceptors.i1.preserveExisting=falsea2.sinks.k1.type = avro a2.sinks.k1.hostname = 127.0.0.1a2.sinks.k1.port = 44444 a2.sinks.k1.channel = c1# Use a channel which buffers events in memorya2.channels.c1.type = memorya2.channels.c1.capacity = 1000a2.channels.c1.transactionCapacity = 100a2.sources.r1.type = execa2.sources.r1.shell = /bin/bash -ca2.sources.r1.command = tail -f /var/log/nginx_2/access_`date +%Y%m%d%H`.loga2.sources.r1.channels = c1####################################server配置[root@host_12 test]# cat h1.conf# Name the components on this agentserv_1.sources = r1serv_1.sinks = k2 k3 serv_1.channels = c2 c3#serv_1.sources.r1.selector.type = replicatingserv_1.sources.r1.selector.type = multiplexingserv_1.sources.r1.selector.header = nginxserv_1.sources.r1.selector.mapping.nginx_1 = c2serv_1.sources.r1.selector.mapping.nginx_2 = c3 serv_1.sources.r1.type = avro serv_1.sources.r1.channels = c2 c3 serv_1.sources.r1.bind = 0.0.0.0 serv_1.sources.r1.port = 44444 serv_1.channels.c2.type=memory serv_1.channels.c2.capacity=2000000 serv_1.channels.c2.transactionCapacity=10000 serv_1.channels.c3.type=memory serv_1.channels.c3.capacity=2000000 serv_1.channels.c3.transactionCapacity=10000 serv_1.sinks.k2.channel = c2serv_1.sinks.k2.type = file_rollserv_1.sinks.k2.batchSize = 100000000serv_1.sinks.k2.rollInterval = 1000000serv_1.sinks.k2.serializer = TEXTserv_1.sinks.k2.sink.directory = /var/log/flume/ serv_1.sinks.k3.channel = c3 serv_1.sinks.k3.type = logger