服务监控（技术框架Flume、Kafka、Storm、Hadoo

主要涉及到的技术框架：flume（日志收集及传输）、kafka（消息队列）、storm（流式计算）、hadoop（离线分析），这几项技术也是大数据方面较为成熟和常用的技术方案。大数据是未来的一个热点方向，涉及的技术和思想也十分丰富。本文仅对此次服务监控中应用到的主要技术框架的基本原理和核心概念做一介绍

Flume：

1. 简介：

Flume ng是cloudera提供的一个分布式、可靠、高可用的系统，它能够将不同数据源的海量日志数据进行高效收集、聚合、移动，最后存储到一个数据存储系统中（flume 的核心就是把数据从数据源收集过来，再送到目的地）。由原来的flume og到现在的flume ng，进行了架构重构。改动的另一原因是将 flume 纳入 apache 旗下，cloudera flume 改名为 apache flume。

2. 基本架构：

 Event：

代表着一个数据流的最小完整单元，通过source从外部数据源输入，流经channel，最终通过sink向外部输出，比如我们日志中的一条日志

 Flow：

Event从源点到达目的点的迁移的抽象

 Agent：

一个独立的flume进程，包含三个核心组件source、channel、sink，通过这些组件，event可以从一个地方流向下一个地方

 Source：

用来消费传递到该组件的event，可以接收外部源发送过来的数据。不同的 source可以接受不同的数据格式。flume支持avro、exec、spool、http等source类型

1. Exec source：以运行 linux 命令的方式，持续的输出最新的数据，如 tail -f 指令。 exec source 可以实现对日志的实时收集，但是存在flume不运行或者指令执行出错时，将无法收集到日志数据，无法保证日志数据的完整性

2. Spooling directory source：监测配置的目录下新增的文件，并将文件中的数据读取出来。spool source 虽然无法实现实时的收集数据，但是可以使用以分钟的方式分割文件，趋近于实时。

更多其他的source类型参考官方文档http://flume.apache.org/flumeuserguide.html#flume-sources

 Channel：

中转event的一个临时存储，保存有source组件传递过来的event，可以理解为一个队列。flume支持memory、jdbc、file、kafka等channel类型

1. Memory channel：可以实现高速的吞吐，但是无法保证数据的完整性，如果 java 进程死掉，任何存储在内存的事件将会丢失。另外，内存的空间受到分配内存大小的限制

2. File channel：是一个持久化的channel，它持久化所有的event，并将其存储到磁盘中。因此，即使 java 虚拟机当掉，也不会造成数据丢失

3. Kafka channel（1.6版本新增）：event被存储在kafka集群中，速度慢于memory channel，但是高于file channel，并且可靠性较高。另外此种方式相比之前的source->channel->sink->kafka也更加便捷

更多其他的channel类型参考官方文档http://flume.apache.org/flumeuserguide.html#flume-channels

 Sink：

从channel中读取并移除event，将event传递到下一个节点，可以是storage，也可以是下一个agent（如果有的话）。flume支持hdfs、logger、file roll、elastic search、kafka等sink类型

更多其他的sink类型参考官方文档http://flume.apache.org/flumeuserguide.html#flume-sinks

Kafka：

1. 简介：

Kafka是由linkedin开发的一个分布式的消息（发布／订阅）系统，使用scala编写，作为多种类型的数据管道和消息系统使用。主要特性如下：

 以时间复杂度为o(1)的方式提供消息持久化能力，即使对tb级以上数据也能保证常数时间复杂度的访问性能（得益于其对数据文件的顺序写入（append））

 高吞吐率。即使在非常廉价的商用机器上也能做到单机支持每秒100k条以上消息的传输

 支持kafka server间的消息分区，及分布式消费，同时保证每个partition内的消息顺序传输

 同时支持离线数据处理和实时数据处理（group机制）

 Scale out：支持在线水平扩展

2. 基本架构：

作为一个订阅／发布系统，kafka最基本的结构类似上图，主要包含三类角色：producer（消息发布（数据写入））、kafka cluster（数据存储）、consumer（订阅消费（数据读取））。更为全面的拓扑结构类似下图：

一个典型的kafka集群中包含若干producer，若干broker（kafka集群，支持水平扩展），若干consumer group，以及一个zookeeper集群。kafka通过zookeeper管理集群配置。producer使用push模式将消息发布到broker，consumer使用pull模式从broker订阅并消费消息

3. 核心概念：

 Topic：

一个topic可以认为是一类消息，topic在逻辑上可以被认为是一个queue，每条消息都必须指定它所属的topic，可以简单理解为必须指明把这条消息放进哪个queue里，consumer也要指明消费哪个topic的数据

 Partition：

1. 每个topic将被分成一个或多个partition，每个partition在物理上对应一个目录，该目录下存储这个partition的所有消息文件和索引文件。任何发布到此partition的消息都会被直接append到log文件的尾部，属于顺序写磁盘，因此效率非常高，这是kafka高吞吐率的一个很重要的保证，逻辑结构如下图：

2. Partitions的设计目的有多个：最根本原因是kafka基于文件存储，通过分区，可以将日志内容分散到多个kafka实例（broker）上，来避免文件尺寸达到单机磁盘的上限,也能分散读写压力；此外越多的partition意味着可以容纳更多的consumer,有效提升并发消费的能力（原因见下文）

 Log：

Kafka底层通过文件系统存储数据，这里所说的文件在kafka里称作log，如下图：

1. 日志文件中保存了一序列“log entries”（日志条目）。每个日志都有一个offset来唯一的标记一条消息，每个partition在物理存储层面有多个log file组成（称为segment）。当segment文件尺寸达到一定阀值时，将会创建一个新的文件（控制单个文件大小）。当buffer中消息的条数达到阀值时将会触发数据flush到数据文件中，同时如果“距离最近一次flush的时间差”达到阈值时也会触发flush到日志文件（批量写入，避免频繁io操作）

2. 对于传统的mq而言，一般会删除已经被消费的消息，而kafka集群会保留所有的消息，无论其是否被消费（这样做的一点好处是当有需要的时候可以重复消费已消费过的数据）。当然，因为磁盘限制，不可能永久保留所有数据（实际上也没必要），因此kafka提供两种策略删除旧数据。一是基于时间，二是基于partition文件大小

 Producer：

Producer将消息发布到指定的topic中，同时producer也能通过paritition策略选择将其存储到哪一个partition，如果一个topic只有一个partition，那这个partition所在的机器的io将有可能成为这个topic的性能瓶颈，而有了多partition后，不同的消息可以并行写入不同broker的不同partition里，极大的提高了吞吐率

 Consumer group：

1. 本质上kafka只支持topic，每个consumer属于一个consumer group。反过来说，每个group中可以有多个consumer。发送到topic的一条消息，只会被订阅此topic的每个group中的一个consumer消费，如下图：

2. 这是kafka用来实现一个topic消息的广播（发给所有的consumer group）和单播（发给某一个consumer group）的手段。多个consumer group可以订阅同一个topic。如果需要实现广播，只要consumer归属于不同的group就可以了（同一份数据采用不同的处理方式，比如实时计算和离线存储）。要实现单播只要所有的consumer在同一个group里

 Push vs pull

在kafka中，采用了pull方式，即consumer在和broker建立连接之后，主动去pull消息。这种模式有其优点，consumer端可以根据自己的消费能力适时的去pull消息并处理，且可以控制消息消费的进度，消费者可以良好的控制消息消费的数量

 核心概念总结：

1. 一个partition中的消息只会被group中的一个consumer消费

2. 每个group中consumer消费互相独立

3. 我们可以认为一个group是一个“订阅”者，一个topic中的每个partition，只会被一个group中的一个consumer消费，不过一个consumer可以消费多个partition中的消息

4. Kafka只能保证一个partition中的消息被某个consumer消费时是有序的

 Replication：

1. Kafka将每个partition数据复制到多个server上，每个partition有一个leader和多个follower

2. Leader处理所有的读写请求，follower需要和leader保持同步。leader负责跟踪所有follower的状态，如果follower"落后"太多或者断开连接，leader将会把它从同步列表中删除

3. 当所有的follower都将一条消息保存成功，此消息才被认为是"committed"，那么此时consumer才能消费它

4. 即使只有一个实例存活，仍然可以保证消息的正常发送和接收，只要zookeeper集群存活即可

Storm：

1. 简介：

Storm是一个分布式的、高容错的实时计算系统。类似于hadoop的map和reduce原语，storm提供了spout和bolt原语，使得实时计算的编程模型变得非常简单。目前主要的应用场景包括：实时分析（计算）、在线机器学习等

2. 基本架构：

 Nimbus：负责资源分配和任务调度

 Supervisor：负责接受nimbus分配的任务，启动和停止属于自己管理的worker进程

 Worker：运行具体处理组件逻辑的进程

 Task：worker中运行的具体的组件（spout/bolt）

数据流：

太抽象？再来张具体点儿的！

图中可以看到一些词汇：spout（数据来源），bolt（数据处理），stream grouping（通过分组决定数据具体流向），而这一切最终绘制出一幅图（topology），上图展示了storm中的数据流，也看到了其中最核心的一些组件，下面具体介绍

3. 核心概念：

 Topology：

一个实时计算应用程序的逻辑在storm里面被封装到topology对象里面， 翻译过来可以叫做计算拓补。storm里面的topology相当于hadoop里面的一个mapreduce job，它们的关键区别是：一个mapreduce job最终总是会结束的， 然而一个storm的topology会一直运行，除非你显式的kill。一个topology是spouts和bolts组成的图状结构，而连接spouts和bolts的则是stream groupings

 Tuple：

一次消息传递的基本单元，本来应该是一个key-value的map，但是由于各个组件间传递的tuple的字段名称已经事先定义好，所以tuple中只要按序填入各个value就行了，所以就是一个value list

 Stream：

Stream是storm里面的关键抽象。一个stream是一个没有边界的tuple序列，也就是流式计算里的流

 Spout：

在一个topology中产生源数据流的组件。通常情况下spout会从外部数据源中读取数据，然后转换为topology内部的源数据。spout接口中有个nexttuple()方法，storm框架会不停地调用此函数，用户只要在其中生成源数据即可

 Bolt：

在一个topology中接受数据然后执行处理的组件。bolt可以执行过滤、合并、写数据库等任何操作。bolt接口中有个execute(tuple input) 方法,在接受到数据后会调用此方法，用户可以在其中执行自己想要的操作

 Stream grouping：

定义了一个流在bolt任务间该如何被切分，说白了就是谁来处理哪些数据，数据该流向哪里（消息分发策略），下面列举几种storm中常用的stream grouping类型:

1. Shuffle grouping：随机分组，随机派发stream里面的tuple给后续的bolt，保证每个bolt接收到的tuple数目基本相同

2. Fields grouping：按字段分组，比如按userid来分组，具有同样userid的tuple会被分发到相同的bolt，而不同的userid则会被分配到不同的bolt

3. All grouping：广播发送，对于每一个tuple，所有的bolt都会收到

4. Global grouping：全局分组，tuple被分配到storm中的一个bolt，再具体一点就是分配给id值最低的那个task

更多grouping方式详见：http://storm.apache.org/documentation/concepts.html

 并发度：

影响storm topology并发度的因素（遇到性能问题时通过这三个方面进行调整）：

1. Worker进程数

2. Task线程数

3. Task数

下面看个具体的例子：

// use two worker processes

Conf.setnumworkers(2);

// set parallelism hint to 2

Topologybuilder.setspout("blue-spout", new bluespout(), 2);

Topologybuilder.setbolt("green-bolt", new greenbolt(), 2).setnumtasks(4);

Topologybuilder.setbolt("yellow-bolt", new yellowbolt(), 6);

Hadoop：

1. 简介：

Hadoop是一个开源的可运行于大规模集群上的分布式文件系统和运行处理基础框架，擅长于在廉价机器搭建的集群上进行海量数据的存储与离线处理，具有可靠、高效、可伸缩的特点。Hadoop是一个生态系统，其下包含了很多相关技术组件和框架，主要包括hdfs（分布式文件系统）、mapreduce（分布式计算框架）、hive（基于hadoop的数据仓库）、hbase（分布式列存数据库）zookeeper（分布式协作服务）、avro（序列化系统）等。今天我们主要介绍一下我们在实际使用中接触较多的mapreduce

2. 基本架构：

那么究竟什么是mapreduce？举个简单的例子来说明，你想数出一摞牌中有多少张黑桃，直观方式是自己一张一张检查并且数出有多少张是黑桃。mapreduce方法则是：

 把牌分给其他几个人

 让每个人数自己手中的牌有几张是黑桃，然后把这个数字汇报给你（通过把牌分给多个玩家并且让他们各自数数，你就在并行执行运算，因为每个玩家都在同时计数，这同时把这项工作变成了分布式的）——map

 你把所有玩家告诉你的数字加起来，得到最后的结论——reduce

当然这只是打个比方，实际上并不严谨，我们来看下面的图：

Mapreduce可以理解为是一套编程模型，更具体一点可以理解为一个过程。在这个过程中主要包括了以下几个阶段：map、shuffle（分map端及reduce端）、reduce，下面通过wordcount（假设我们有n篇文章，我们需要对这n篇文章中出现的单词分别统计出现的次数）为例（此例被誉为hadoop上的hello world）分别对这几个阶段作一介绍

3. 关键步骤：

对读取的语句进行map操作，输出为形式，其中key为单词，value为单词出现1次

Shuffle：

Shuffle要解决的问题就是怎样把map的输出结果有效地传送到reduce端。也可以这样理解， shuffle描述着数据从map端输出到reduce端输入的这段过程。这段过程在map端大致包含partition、spill、merge，在reduce端包含copy、merge

Map端：

1.Partition：

假设将来我们的job有3个reduce task，那么就要决定map之后的数据将来交由哪个reduce后续处理，这就是partition阶段做的工作Mapreduce提供partitioner接口，它的作用就是根据key及reduce的数量来决定当前的这对输出数据最终应该交由哪个reduce task处理。默认对key hash后再以reduce task数量取模。默认的取模方式只是为了平均reduce的处理能力，如果用户自己对partition有需求，可以自行定制

2.Spill：

接下来，需要将数据写入内存缓冲区中，缓冲区的作用是批量收集map结果，减少磁盘io的影响。当缓冲区使用率到达一个阈值（100*0.8），就会将缓冲区数据写入磁盘，这个过程就是spill，中文可以翻译为溢写，很形象也很好理解。spill过程会启动一个新的线程，不会影响map task继续输出结果。spill过程中会对将要写入磁盘的数据按照key进行排序（sort）在wordcount事例中， map的输出可以看到有很多类似、、、、这样相同的，如果把这些全部写入磁盘，第一浪费空间，第二增加磁盘io。所以在这里有一个潜在的优化点，就是把key相同的数据进行合并，合并后的数据类似、，这个过程叫做combine。combine过程是可选的，我们可以在job中设置相应的combiner。combiner有助于优化中间结果，但并不是所有场景都适合使用，一个总的原则就是，combiner绝不能影响最终的计算结果（累加这种场景就可以使用）

3.Merge：

每一次spill都会产生一个溢写文件，当数据量使得spill多次发生时，就会产生多个溢写文件。而最终当map过程完成的时候，需要将这些溢写文件合并为一个，这个过程就是merge在这个过程中可能会遇到类似这样的数据、、、，因为他们具有相同的key，所以会被合并，结果类似、至此，map端的shuffle过程就完成了，下面再看一下reduce端的shuffle过程

Reduce端：

1. Copy：

在上一阶段，map端生成了作为reduce端输入的数据文件，在reduce真正运行之前，所有的时间都是在拉取数据，做merge。copy过程，就是reduce端启动一些线程，从map端拉取数据的过程

2. Merge：

Copy过来的数据首先会存入内存缓冲区，这里缓冲区的大小要比map端更为灵活，它基于jvm的heap。因为在shuffle阶段，reduce还没有开始运行，所以可以把更多的内存分配给shuffle过程来使用。类似于map端，当内存缓冲区达到一定阈值后，将数据写入磁盘，并最终生成一个文件，这个文件就是reduce端数据输出的文件

至此，整个shuffle过程就完成了

 Reduce：

Reduce的输入为，经过reduce汇总（遍历累计合并结果），输出，key为单词，value为单词出现的总次数