针对于系统集群节点的管理，需要通过系统集群节点的底层获取日志数据来实现对集群节点的管理和监控，将集群节点的数据进行采集，通过利用大数据分析系统和神经网络的算法，实现对集群节点的预测以及系统扩容，缩容等业务的扩展。在整个系统中日志数据的采集对整个系统至关重要。因此本文讲详细介绍有关业界的常见的日志数据采集架构原理。通同时讲在个人github上建立一套完整的日志采集系统，供大家参考和学习，同时讲详细的讲述个人项目的整体架构和实现的原理。github网址为：GitHub - 2462612540/Cluster-Manager: To cluster node in system management, through the system cluster nodes the underlying access log data to implement the management of the cluster nodes and monitoring, the cluster node of the data collection, through the use of big data analysis system and neural network algorithm, realize the prediction of the cluster nodes and system expansion, shrinkage and business extension. The collection of log data is very important for the whole system. Therefore, this article describes in detail the common log data collection architecture in the industry. At the same time, I will build a complete log collection system on github for your reference and study. At the same time, I will talk about the overall architecture and implementation principle of my personal project in detail.
当前主流常用海量数据采集工具 当前主流的数据采集系统构架原理 美团的基于Flume的日志收集系统架构设计日志收集是大数据的基石。许多公司的业务平台每天都会产生大量的日志数据。收集业务日志数据，供离线和在线的分析系统使用，正是日志收集系统的要做的事情。高可用性，高可靠性和可扩展性是日志收集系统所具有的基本特征。目前常用的开源日志收集系统有Flume, Scribe等。Flume是Cloudera提供的一个高可用的，高可靠的，分布式的海量日志采集、聚合和传输的系统，目前已经是Apache的一个子项目。Scribe是Facebook开源的日志收集系统，它为日志的分布式收集，统一处理提供一个可扩展的，高容错的简单方案。
常用的开源日志收集系统对比
美团的日志收集系统负责美团的所有业务日志的收集，并分别给Hadoop平台提供离线数据和Storm平台提供实时数据流。美团的日志收集系统基于Flume设计和搭建而成。目前每天收集和处理约T级别的日志数据。
a. 整个系统分为三层：Agent层，Collector层和Store层。其中Agent层每个机器部署一个进程，负责对单机的日志收集工作；Collector层部署在中心服务器上，负责接收Agent层发送的日志，并且将日志根据路由规则写到相应的Store层中；Store层负责提供永久或者临时的日志存储服务，或者将日志流导向其它服务器。
b. Agent到Collector使用LoadBalance策略，将所有的日志均衡地发到所有的Collector上，达到负载均衡的目标，同时并处理单个Collector失效的问题。
c. Collector层的目标主要有三个：SinkHdfs, SinkKafka和SinkBypass。分别提供离线的数据到Hdfs，和提供实时的日志流到Kafka和Bypass。其中SinkHdfs又根据日志量的大小分为SinkHdfs_b，SinkHdfs_m和SinkHdfs_s三个Sink，以提高写入到Hdfs的性能，具体见后面介绍。
d. 对于Store来说，Hdfs负责永久地存储所有日志；Kafka存储最新的7天日志，并给Storm系统提供实时日志流；Bypass负责给其它服务器和应用提供实时日志流。
美团的日志收集系统的模块分解详解Agent, Collector和Bypass中的Source, Channel和Sink的关系。
- a. 模块命名规则：所有的Source以src开头，所有的Channel以ch开头，所有的Sink以sink开头；
- b. Channel统一使用美团开发的DualChannel，具体原因后面详述；对于过滤掉的日志使用NullChannel，具体原因后面详述；
- c. 模块之间内部通信统一使用Avro接口；
- 可用性(availablity)
对日志收集系统来说，可用性(availablity)指固定周期内系统无故障运行总时间。要想提高系统的可用性，就需要消除系统的单点，提高系统的冗余度。下面来看看美团的日志收集系统在可用性方面的考虑。
Agent死掉
Agent死掉分为两种情况：机器死机或者Agent进程死掉。
对于机器死机的情况来说，由于产生日志的进程也同样会死掉，所以不会再产生新的日志，不存在不提供服务的情况。
对于Agent进程死掉的情况来说，确实会降低系统的可用性。对此，我们有下面三种方式来提高系统的可用性。首先，所有的Agent在supervise的方式下启动，如果进程死掉会被系统立即重启，以提供服务。其次，对所有的Agent进行存活监控，发现Agent死掉立即报警。最后，对于非常重要的日志，建议应用直接将日志写磁盘，Agent使用spooldir的方式获得最新的日志。
Collector死掉
由于中心服务器提供的是对等的且无差别的服务，且Agent访问Collector做了LoadBalance和重试机制。所以当某个Collector无法提供服务时，Agent的重试策略会将数据发送到其它可用的Collector上面。所以整个服务不受影响。
Hdfs正常停机
我们在Collector的HdfsSink中提供了开关选项，可以控制Collector停止写Hdfs，并且将所有的events缓存到FileChannel的功能。
Hdfs异常停机或不可访问
假如Hdfs异常停机或不可访问，此时Collector无法写Hdfs。由于我们使用DualChannel，Collector可以将所收到的events缓存到FileChannel，保存在磁盘上，继续提供服务。当Hdfs恢复服务以后，再将FileChannel中缓存的events再发送到Hdfs上。这种机制类似于Scribe，可以提供较好的容错性。
Collector变慢或者Agent/Collector网络变慢
如果Collector处理速度变慢（比如机器load过高）或者Agent/Collector之间的网络变慢，可能导致Agent发送到Collector的速度变慢。同样的，对于此种情况，我们在Agent端使用DualChannel，Agent可以将收到的events缓存到FileChannel，保存在磁盘上，继续提供服务。当Collector恢复服务以后，再将FileChannel中缓存的events再发送给Collector。
Hdfs变慢
当Hadoop上的任务较多且有大量的读写操作时，Hdfs的读写数据往往变的很慢。由于每天，每周都有高峰使用期，所以这种情况非常普遍。
对于Hdfs变慢的问题，我们同样使用DualChannel来解决。当Hdfs写入较快时，所有的events只经过MemChannel传递数据，减少磁盘IO，获得较高性能。当Hdfs写入较慢时，所有的events只经过FileChannel传递数据，有一个较大的数据缓存空间。
- 可靠性(reliability)
对日志收集系统来说，可靠性(reliability)是指Flume在数据流的传输过程中，保证events的可靠传递。
对Flume来说，所有的events都被保存在Agent的Channel中，然后被发送到数据流中的下一个Agent或者最终的存储服务中。那么一个Agent的Channel中的events什么时候被删除呢？当且仅当它们被保存到下一个Agent的Channel中或者被保存到最终的存储服务中。这就是Flume提供数据流中点到点的可靠性保证的最基本的单跳消息传递语义。
那么Flume是如何做到上述最基本的消息传递语义呢？
首先，Agent间的事务交换。Flume使用事务的办法来保证event的可靠传递。Source和Sink分别被封装在事务中，这些事务由保存event的存储提供或者由Channel提供。这就保证了event在数据流的点对点传输中是可靠的。在多级数据流中，如下图，上一级的Sink和下一级的Source都被包含在事务中，保证数据可靠地从一个Channel到另一个Channel转移。
其次，数据流中 Channel的持久性。Flume中MemoryChannel是可能丢失数据的（当Agent死掉时），而FileChannel是持久性的，提供类似mysql的日志机制，保证数据不丢失。
-
可扩展性(scalability)
对日志收集系统来说，可扩展性(scalability)是指系统能够线性扩展。当日志量增大时，系统能够以简单的增加机器来达到线性扩容的目的。
对于基于Flume的日志收集系统来说，需要在设计的每一层，都可以做到线性扩展地提供服务。下面将对每一层的可扩展性做相应的说明。
Agent层
对于Agent这一层来说，每个机器部署一个Agent，可以水平扩展，不受限制。一个方面，Agent收集日志的能力受限于机器的性能，正常情况下一个Agent可以为单机提供足够服务。另一方面，如果机器比较多，可能受限于后端Collector提供的服务，但Agent到Collector是有Load Balance机制，使得Collector可以线性扩展提高能力。
Collector层
对于Collector这一层，Agent到Collector是有Load Balance机制，并且Collector提供无差别服务，所以可以线性扩展。其性能主要受限于Store层提供的能力。
Store层
对于Store这一层来说，Hdfs和Kafka都是分布式系统，可以做到线性扩展。Bypass属于临时的应用，只对应于某一类日志，性能不是瓶颈。
-
Channel的选择
Flume1.4.0中，其官方提供常用的MemoryChannel和FileChannel供大家选择。其优劣如下：
- MemoryChannel: 所有的events被保存在内存中。优点是高吞吐。缺点是容量有限并且Agent死掉时会丢失内存中的数据。
- FileChannel: 所有的events被保存在文件中。优点是容量较大且死掉时数据可恢复。缺点是速度较慢。
上述两种Channel，优缺点相反，分别有自己适合的场景。然而，对于大部分应用来说，我们希望Channel可以同提供高吞吐和大缓存。基于此，我们开发了DualChannel。
- DualChannel：基于 MemoryChannel和 FileChannel开发。当堆积在Channel中的events数小于阈值时，所有的events被保存在MemoryChannel中，Sink从MemoryChannel中读取数据； 当堆积在Channel中的events数大于阈值时， 所有的events被自动存放在FileChannel中，Sink从FileChannel中读取数据。这样当系统正常运行时，我们可以使用MemoryChannel的高吞吐特性；当系统有异常时，我们可以利用FileChannel的大缓存的特性。
- 和scribe兼容
在设计之初，我们就要求每类日志都有一个category相对应，并且Flume的Agent提供AvroSource和ScribeSource两种服务。这将保持和之前的Scribe相对应，减少业务的更改成本。
- 权限控制
在目前的日志收集系统中，我们只使用最简单的权限控制。只有设定的category才可以进入到存储系统。所以目前的权限控制就是category过滤。
如果权限控制放在Agent端，优势是可以较好地控制垃圾数据在系统中流转。但劣势是配置修改麻烦，每增加一个日志就需要重启或者重载Agent的配置。
如果权限控制放在Collector端，优势是方便进行配置的修改和加载。劣势是部分没有注册的数据可能在Agent/Collector之间传输。
考虑到Agent/Collector之间的日志传输并非系统瓶颈，且目前日志收集属内部系统，安全问题属于次要问题，所以选择采用Collector端控制。
-
提供实时流
美团的部分业务，如实时推荐，反爬虫服务等服务，需要处理实时的数据流。因此我们希望Flume能够导出一份实时流给Kafka/Storm系统。
一个非常重要的要求是实时数据流不应该受到其它Sink的速度影响，保证实时数据流的速度。这一点，我们是通过Collector中设置不同的Channel进行隔离，并且DualChannel的大容量保证了日志的处理不受Sink的影响
-
系统监控
对于一个大型复杂系统来说，监控是必不可少的部分。设计合理的监控，可以对异常情况及时发现，只要有一部手机，就可以知道系统是否正常运作。对于美团的日志收集系统，我们建立了多维度的监控，防止未知的异常发生。
发送速度，拥堵情况，写Hdfs速度
通过发送给zabbix的数据，我们可以绘制出发送数量、拥堵情况和写Hdfs速度的图表，对于超预期的拥堵，我们会报警出来查找原因。下面是Flume Collector HdfsSink写数据到Hdfs的速度截图：
下面是Flume Collector的FileChannel中拥堵的events数据量截图：
- flume写hfds状态的监控
Flume写入Hdfs会先生成tmp文件，对于特别重要的日志，我们会每15分钟左右检查一下各个Collector是否都产生了tmp文件，对于没有正常产生tmp文件的Collector和日志我们需要检查是否有异常。这样可以及时发现Flume和日志的异常.
-
日志大小异常监控
对于重要的日志，我们会每个小时都监控日志大小周同比是否有较大波动，并给予提醒，这个报警有效的发现了异常的日志，且多次发现了应用方日志发送的异常，及时给予了对方反馈，帮助他们及早修复自身系统的异常。
通过上述的讲解，我们可以看到，基于Flume的美团日志收集系统已经是具备高可用性，高可靠性，可扩展等特性的分布式服务
Flume定制实战——日志平台架构解析 项目介绍公司主导开发【统一日志平台】时采用的技术(主要技术栈：flume+ES+Redis+mongoBD+Kafka+Hadoop+Netty )， 日志平台数据量为8亿/天，高峰为8500万/小时、800万/5分钟。 flume agent单机压测15000/s数据量，未出现程序异常、资源占用过高与日志明显丢失情况。
背景说明我们的需求是将Java 应用的log信息进行收集，达到日志采集的目的，agent目前主要有flume、Logstash，技术选型详情在此就不表了，最终选择的flume。
由于当时公司内部推行技术组件一直有难度，且也无法借助行政手段，因此我们在设计时很多时候考虑都是尽量对应用透明，比如我们的flume source使用的是基于log文件的，而未使用应用与flume agent采用长连接的方式（该方式需要修改log4j配置，并且引入我们的jar），比如我们agent进行日志等级判断时 需要兼容各种日志格式，因为我们难以推动各个应用方统一日志格式……
sre方面，当时并没有为agent预留内存等资源，所以一旦我们的agent出现资源占用过多，都会比较敏感。
可以看到我们使用kafka将log信息做转储，消息消费者主要有HDFS、ES、Queue等。
定制开发source定制：dirSource
基于文件的dirSource在flume高版本里已经去除了，原生的dirSource也存在很多性能和功能上的问题，为了在我们使用的flume1.6版本里继续使用dirSource，我们就基于1.6实现了一版dirSource。
dirSource特性
- 基于NIO的WatchEvent进行log文件内容的写操作监听，同时有能动态的监听文件的创建和删除。我们丰富了这部分的匹配模式，可以实现灵活的文件监听。
- 文件的读取基于RandomAccessFile，按行读取
- 将获取内容进行处理封装Event，存入Channel。
存在的问题
- 无论是WatchEvent还是RandomAccessFile在log疯狂输出时，CPU占用会居高不下。
execSource
execSource为flume新版本推出的用来替代dirSource的一种实现方式，主要是通过Java执行shell命令，并且获取shell命令的输出信息，如tail、cat等。
我们在原生的execSource基础上，实现了文件的自动监听，实现了多命令模式，并且会自动回收长时间无内容产出的命令，优化了原有的线程关闭的操作及进程钩子等。
execSource特性
- 基于NIO的WatchEvent进行log文件内容的写操作监听，同时有能动态的监听文件的创建和删除。我们丰富了这部分的匹配模式，可以实现灵活的文件监听
- 多命令模式
- 自动回收长时间无内容产出的命令
- 重启时自动清理无用的shell命令
存在的问题
- flume agent进程被kill -9 时，对导致执行的shell命令无法退出，进而导致句柄得不到释放，积累下来对服务器造成影响。
我们采用的是kafka sink，flume原生的kafka sink使用的是老版本kafka producer client，发送消息时需要手动实现批量与异步，并且是消息发送的实现上存在一些不足，在大数据量时存在明显的性能瓶颈，并且会由于集合中消息数量太多而报异常，进而丢失消息。我们定制的版本使用的new kafka producer client ，并且对消息发送做了优化，同时对Channel参数做了大量的压测，最终确定了最优配置。
kafkaSink特性
- 使用new kafka producer client ，默认异步批量发送
- 优化了消息体序列化方式
测试一
下文描述的压测都是在建设日志平台过程中对flume的相关测试。 测试环境都是mac book pro ，这里只关注各个测试项的对比信息。
说明：
- 类型New kafka sink为：原生sink，使用kafka旧client，只定制了从head中获取配置参数，拼接字符串
- 类型Old kafka sink为：深度定制版，使用kafka新client
结论：
- flume 资源占用从kafka发送部分目前没有太好的优化方案，且旧kafka client数据丢失更加严重。
- 因此flume kafka sink 维持不变，后续可从flume source入手优化
测试二
测试三
配置说明一
a1.sinks.k1.batch.num.messages = 5000
a1.sinks.k1.block.on.buffer.full = true
a1.sinks.k1.buffer.memory = 167108864
a1.channels.c1.type = memory
a1.channels.c1.capacity = 100000
a1.channels.c1.transactionCapacity = 1000
flume -Xmx256M -Xms256M测试结果一
日志写数量
用时
线程数
QPS
日志文件量
成功发送到kafka数量
topic个数
CPU
内存
序列化方式
其他
500万
74s
50
70000/s
600m
280万（单个topic）
2
未统计
300M
fastjson
agent异常
配置说明二
a1.sinks.k1.batch.num.messages = 5000
a1.sinks.k1.block.on.buffer.full = true
a1.sinks.k1.buffer.memory = 167108864
a1.channels.c1.type = memory
a1.channels.c1.capacity = 500000
a1.channels.c1.transactionCapacity = 500
a1.channels.c1.byteCapacity = 536870912
flume -Xmx256M -Xms256M 测试结果二
日志写数量
用时
线程数
QPS
日志文件量
成功发送到kafka数量
topic个数
CPU
内存
序列化方式
其他
500万
68s
50
74000/s
600m
500万（单个topic）
2
200%以上
320M
fastjson
无异常
500万
68s
50
74000/s
600m
500万（单个topic）
1
100%-200%
320M
fastjson
无异常
500万
68s
50
74000/s
600m
500万（单个topic）
1
小于100%
280M
StringBuild拼接
无异常
总结- 数据量过大时，sink中kafka client 缓存被存满，kafka会报异常，设置block=true后，存入缓存会被阻塞，kafka不报异常，但是由于sink从channel中消费的速度远低于source存入channel的速度，channel会报Unable to put event on required channel，flume停止提供服务。继续写入日志，会重复发送错误。
- 该异常可通过增大channel的byteCapacity参数或者调大JVM的参数值（byteCapacity默认为JVM的80%）来提高报错的阀值，且减小transactionCapacity 的值来减缓传输到sink的数据量。
- JVM内存参数在7万每秒的压力下，设置为256M较为合适，byteCapacity设置为512M较为合适，当增加channel个数或者增大channel向sink传输的数据量时，都会导致sink消费过慢报异常（总结1中异常），单个channel内存消耗在300M左右。
- 对于数据量较大的应用，建议只发送单个topic。
高可用Hadoop平台－Flume NG实战图解篇-阿里云开发者社区阿里云的高可用日志设计架构
基于Flume的美团日志收集系统(一)架构和设计 - 美团技术团队美团的日志收集系统架构
Flume定制实战——日志平台架构解析 - 云+社区 - 腾讯云 Flume的日志采集系统实战
有赞百亿级日志系统架构设计 - 知乎 有赞百亿级日志系统架构设计
日志采集框架Flume - java与大数据征程 - 博客园
六大主流大数据采集平台架构分析_Jason的专栏-CSDN博客
百亿级日志系统架构设计及优化_架构文摘-CSDN博客
大数据日志采集分析框架
