5.大数据生态圈成员和原理_aggregationtype 为 replace

11有界数据与无界数据
知识犹如人体的血液一样宝贵。
——高士其
上一章我们学习了人工智能下的大数据，这一章我们会从原理、架构角度深入学习大数据生态系统各个组件。

为了给后面的小节做铺垫，这一小节我们来学习一下有界数据和无界数据这两个概念。

在前面章节我们讲过，大数据是具有4V 特征：volume（容量）、variety（种类）、velocity（速度）、value（价值）。但是不管是怎样的数据，按照时间上分割的话只有两类：有界数据、无界数据。

有界数据：数据的开始和结束都有明确的定义。

无界数据：数据的开始有定义，结束没有定义。

有界数据场景和无界数据场景

关于有界数据，我们每个人日常生活中会有很多亲身体验。例如中国移动、中国联通、中国电信每个月发送到手机上的账单。

手机卡月账单数据计算的是一个月的第一天到最后一天时间范围内所有接打电话、收发短信、使用流量浏览网络等所有行为产生费用信息。类似的有界数据还有每天晚上微信运动推送到手机上的步数排行榜、余额宝每天的收益信息等等。

在另外一些场景，数据具有很强的连续性。例如：小区里面的监控录像数据、股票交易数据、电商平台后台交易数据。这些数据在一段时间里会一直连续的产生出来，就像水流一样，无穷无尽。像这样的数据就叫做无界数据。

多数时候有界数据和无界数据之前是可以相互转化的。例如：无界的电商的交易数据对于双十一这一天来说就是有界的。用户使用手机的消费数据相对于一个月来说是有界，但是如果从整个使用周期来看又是无界数据。

有界数据和无界数据通过对“时间窗口”进行转化。

无界数据取某个特定的“时间窗口”就变成了有界数据、有界数据去掉“时间窗口”就变成了无界数据。

批处理和流处理

批处理(Batch Processing)：对一批数据进行处理。

流处理(Stream Processing)：对数据流进行处理。

显而易见，批处理对应有界数据，流处理对应无界数据。

流处理需要在更短的内处理完海量的数据，相对于批处理来说有着更高的实时性、更小的延迟性要求。

在大数据刚刚兴起的时候MapReduce 实现了人类对于历史的有界数据批处理能力，人们并没有想过流处理无界数据。

随着MapReduce 不断优化，以及计算速度更快的Spark 计算引擎出现以后，大数据技术对于批处理的需求已经能够很好的满足了。

进而人们开始关注流处理方面的需求。从开始的Storm 到Spark Streamming ，再到Flink 的出现。流处理框架经历了多个版本的进化以后，在数据处理的实时性、吞吐量等关键性能指标上已经发展得比较完善了。

2019 年5G 在中国正式投入商用，全世界各国都将先后进入5G 时代。5G 时代，更多的物联网设备接入到网络中来，智能家居飞入了越来越多的普通家庭。

这些物联网设备、智能家居设备运转过程中会产生海量的无界数据，流处理技术的需求会越来越迫切。

总结

大数据按照时间上划分为：有界数据、无界数据。有界数据的处理叫批处理，无界数据的处理叫流处理。流处理相对于批处理有着更高的实时性、更小的延迟性要求。

大数据处理技术从具备批处理能力开始，发展到流处理能力。随着5G 技术的发展，流处理场景会越来越多，促使大数据流技术向着更高效标准迈进。

12大数据系统架构演进

学习这件事不在乎有没有人教你，最重要的是在于你自己有没有觉悟和恒心。
—— 法布尔

上一小节我们讲到了无界数据和有界数据的概念。这一小节我们来一起学习一下大数据系统架构的演进。

在实际大数据处理场景中，有界数据、无界数据的需求都非常常见，我们在设计大数据系统架构的时候需要考虑兼容这两种数据处理场景。大数据系统架构经历了两代经典架构：Lambda 架构、Kappa 架构。

Lambda架构

著名的大数据实时计算框架Strom 的作者Nathan Marz 提出了Lambda 架构。

对于大数据处理系统Nathan Marz 有着自己的看法。他认为一个设计良好的大数据系统需要具备以下特性：

Robustandfault-tolerant（容错性和鲁棒性）：对大规模分布式系统来说，机器是不可靠的，可能会当机，但是系统需要是健壮、行为正确的，即使是遇到机器错误。

除了机器错误，人更可能会犯错误。在软件开发中难免会有一些Bug，系统必须对有Bug 的程序写入的错误数据有足够的适应能力，所以比机器容错性更加重要的容错性是人为操作容错性。

对于大规模的分布式系统来说，人和机器的错误每天都可能会发生，如何应对人和机器的错误，让系统能够从错误中快速恢复尤其重要。

Lowlatency reads and updates（低延时）：很多应用对于读和写操作的延时要求非常高，要求对更新和查询的响应是低延时的。

Scalable（横向扩容）：当数据量/负载增大时，可扩展性的系统通过增加更多的机器资源来维持性能。也就是常说的系统需要线性可扩展，通常采用scale out（通过增加机器的个数）而不是scale up（通过增强机器的性能）。

General（通用性）：系统需要能够适应广泛的应用，包括金融领域、社交网络、电子商务数据分析等。

Extensible（可扩展）：需要增加新功能、新特性时，可扩展的系统能以最小的开发代价来增加新功能。

Allows ad hoc queries（方便查询）：数据中蕴含有价值，需要能够方便、快速地查询出所需要的数据。

Minimal maintenance（易于维护）：系统要想做到易于维护，其关键是控制其复杂性，越是复杂的系统越容易出错、越难维护。

Debuggable（易调试）：当出问题时，系统需要有足够的信息来调试错误，找到问题的根源。其关键是能够追根溯源到每个数据生成点。

尽管大数据系统需要具备非常多的特性，但是Nathan Marz 对于数据系统的总结非常简单。

数据系统 = 数据 + 查询

基于以上对大数据系统的理解，Nathan Marz 提出Lambda 架构的目标是设计出一个能满足实时大数据系统关键特性的架构，包括有：高容错、低延时和可扩展等。

Lambda 架构整合离线计算和实时计算，融合不可变性（Immunability），读写分离和复杂性隔离等一系列架构原则，可集成Hadoop，Kafka，Storm，Spark，Hbase等各类大数据组件。

http://lambda-architecture.net/

Lambda架构有三个核心模块组成：批处理层（batch layer）、实时处理层（speed layer）、服务层(serving layer)

Lambda架构核心逻辑有以下五点：

（1）所有进入系统的数据，都会被分发到批处理层（batch layer）和快速处理层（speed layer）。

（2）批处理层（batch layer）有两个作用：

​ 管理master的数据（raw数据）：比如用HDFS来存储。

​ 为数据转换为批处理视图做预处理。

（3）服务层（serving layer）用于加载和实现数据库中的批处理视图，以便用户能查询。

（4）快速处理层（speed layer）用于处理新数据和服务层更新造成的高延迟补偿。

（5）任何Query的答案，都能通过合并批处理视图和实时视图的结果来获得。

Lambda 架构是一种通用的架构思想，任何满足以上Lambda 架构核心逻辑的架构都可以归为Lambda 架构。正因为Lambda 架构具备足够的灵活性，在互联网大数据系统架构中被广泛应用。

以下是小米公司的OlAP 大数据系统架构，采用了Lambda 架构。

数据同时被灌入批处理层、快速处理层。批处理层使用Kylin 对历史数据构建多维数据立方体。

快速处理层使用ES 和Kudu 计算实时数据。批处理层、快速处理层的计算结果在服务层合并在一起。

当OLAP 查询通过服务层获取数据结果时，服务层会根据查询时间分别去批处理、快速处理层的结果集里面查询，并将两个数据集进行合并，最终返回OLAP 查询结果。

Lambda 架构通过批处理层、速度处理层的分离实现了拓展性、高容错性，是一种极具价值的大数据架构方案。但是Lambda 架构的缺点也是显而易见的。

为了实现最终Query 查询的无感知，批处理层、速度处理层需要实现两套一样的数据处理逻辑。处理逻辑稍有不同时就会产生错误的数据结果。当数据处理逻辑变复杂的时候，这种逻辑一致性维护压力就非常明显。

同时由于大数据系统本身运维就比较复杂，批处理层、速度处理层两种不同的大数据框架运维起来压力非常大。

Kappa架构

由于Lambda 架构具有显而易见的缺点，Linkedin的Jaykreps 提出了Kappa 架构。

Kappa 架构的核心思想包括以下三点：

用Kafka或者类似的分布式队列系统保存数据，你需要几天的数据量就保存几天。
当需要全量重新计算时，重新起一个流计算实例，从头开始读取数据进行处理，并输出到一个新的结果存储中。
当新的实例做完后，停止老的流计算实例，并把老的一些结果删除。
在Kappa 架构下，只有在有必要的时候才会对历史数据进行重复计算，实时计算和流式计算都是使用同一段代码逻辑，同一计算框架。Kappa 架构中新的计算实例会输出到新表中，即使新的计算逻辑有问题时旧计算实例输出的旧表依然存在，有着较高的容错性。

Lambda架构 VS Kappa架构

Lambda 架构相对于kappa 架构出现的更早，Lambda 架构出现的时期，还没有哪一种大数据计算引擎能同时很好地解决数据批处理和流式处理的问题。

Lambda 架构通过建立批处理层、速度层、服务层的方式，将批处理计算框架、实时计算框架计算的数据结果进行合并，既满足了对历史数据分析的需求，又满足了数据实时性的要求。Lambda 构架是在当时基于有限条件下的一种临时方案，具有历史局限性。

Lambda 架构在代码维护复杂性、计算框架运维复杂性上的问题，Kappa 架构都针对性的进行了改善。然而Kappa 架构也并不是完美的。

因为Kappa 架构对历史数据的计算都是现计算的，当需要追溯的时间跨度很长、计算逻辑复杂的情况下，现计算需要的时间是否能被接受是一个问题。

总结
Lambda 架构是一种早起的大数据系统架构方案，它的出现让人们具备了对历史批量数据、流式数据同时处理的能力。Kappa 架构基于更强计算框架，可以使用同一套代码逻辑、同一套计算框架，对历史批量数据、流式数据统一处理，有着更低的代码维护成本、计算集群运维成本。

13 HDFS：先驱还是先烈

一个不注意小事情的人，永远不会成功大事业。
——戴尔·卡耐基

这一小节我们的话题是HDFS 。带着问题学习的方式往往会比较高效。同学们，我们在学习本小节过程思考一下这个问题：HDFS 是先驱还是先烈？

2004年 Hadoop第一个版本诞生之初就包含了HDFS 和MapReduce 。在此后的15 年里基于Hadoop 的计算框架层出不穷，然而Spark、Flink 这些计算框架想要在Hadoop 生态圈站住脚就必须支持HDFS 作为文件存储系统。可见Hdfs 在Hadoop 生态系统的独特地位。

HDFS技术架构

HDFS 主要包含NameNode 和DataNode。

Namenode 又称为名称节点，是负责管理分布式文件系统的命名空间（Namespace ），保存了两个核心的数据结构，即FsImage 和EditLog 。 你可以把它理解成大管家，它不负责存储具体的数据。

FsImage 用于维护文件系统树以及文件树中所有的文件和文件夹的元数据。

操作日志文件EditLog 中记录了所有针对文件的创建、删除、重命名等操作。

注意，这个两个都是文件，也会加载解析到内存中。

为啥会拆成两个呢? 主要是因为Fsimage 这个文件会很大的，多了之后就不好操作了，就拆分成两个。把后续增量的修改放到EditLog 中, 一个FsImage 和一个Editlog 进行合并会得到一个新的FsImage .

因为它是系统的大管家，如果这个玩意坏了，丢失了怎么办。就相当于你系统的引导区坏了。那就玩完了。整个文件系统就崩溃了。 所以，这个重要的东西，需要备份。

这个时候就产生了一个叫sendaryNamenode的节点用来做备份，它会定期的和Namenode 就行通信来完成整个的备份操作。具体的操作如下:

来源：https://hadoop.apache.org/

SecondaryNameNode的工作情况：

SecondaryNameNode 会定期和NameNode 通信，请求其停止使用EditLog文件，暂时将新的写操作写到一个新的文件edit.new上来，这个操作是瞬间完成，上层写日志的函数完全感觉不到差别；

SecondaryNameNode 通过HTTP GET 方式从NameNode 上获取到FsImage 和EditLog 文件，并下载到本地的相应目录下；

SecondaryNameNode 将下载下来的FsImage 载入到内存，然后一条一条地执行EditLog 文件中的各项更新操作，使得内存中的FsImage 保持最新；这个过程就是EditLog 和FsImage 文件合并；

SecondaryNameNode 执行完（3）操作之后，会通过post方式将新的FsImage 文件发送到NameNode 节点上；

NameNode 将从SecondaryNameNode 接收到的新的FsImage 替换旧的FsImage 文件，同时将edit.new 替换EditLog 文件，通过这个过程EditLog 就变小了；

除了这个自带的备份操作，还需要进行人工的备份，把一份Fsimage 到多个地方进行备份，万一Namenode 的节点坏了呢。

Datanode数据节点，用来具体的存储文件，维护了BlockId 与 Datanode 本地文件的映射。 需要不断地与Namenode 节点通信，来告知其自己的信息，方便Nameode来管控整个系统。

这里还提到一个块的概念，就像Linux本地文件系统中也有块的概念一样，这里也有块的概念。这里的块会默认是128m 每个块都会默认储存三份。

HDFS优缺点

HDFS 之所在Hadoop 生态系统中具有极其重要的地位，是因为具有以下优点：

1、高容错性

数据自动保存多个副本。它通过增加副本的形式，提高容错性。
某一个副本丢失以后，它可以自动恢复，这是由 HDFS 内部机制实现的，我们不必关心。

2、适合批处理

它是通过移动计算而不是移动数据。
它会把数据位置暴露给计算框架。

3、适合大数据处理

处理数据达到 GB、TB、甚至PB 级别的数据。
能够处理百万规模以上的文件数量，数量相当之大。能够处理10K 节点的规模。

4、流式文件访问

一次写入，多次读取。文件一旦写入不能修改，只能追加。
它能保证数据的一致性。

5、可构建在廉价机器上

它通过多副本机制，提高可靠性。
它提供了容错和恢复机制。比如某一个副本丢失，可以通过其它副本来恢复。

同时HDFS也并非完美，有三种场景不适用：

1、低延时数据访问

比如毫秒级的来存储数据，这是不行的，它做不到。
它适合高吞吐率的场景，就是在某一时间内写入大量的数据。但是它在低延时的情况下是不行的，比如毫秒级以内读取数据，这样它是很难做到的。

2、小文件存储

存储大量小文件(这里的小文件是指小于HDFS系统的Block大小的文件（默认64M）)的话，它会占用 NameNode大量的内存来存储文件、目录和块信息。

这样是不可取的，因为NameNode的内存总是有限的。小文件存储的寻道时间会超过读取时间，它违反了HDFS的设计目标。

3、并发写入、文件随机修改

一个文件只能有一个写，不允许多个线程同时写。
仅支持数据 append（追加），不支持文件的随机修改。

HDFS 面临的挑战

HDFS 在2004年刚刚出现的时候，是一个划时代的创举。它让企业第一次获得了将存储大规模数据集存储到廉价计算机上的能力。这种能力奠定了Hadoop开源大数据技术发展的基础，所以说HDFS技术是一项革命性的技术。

HDFS 从出现之日起，就保持快速的性能优化，功能迭代，让它在Hadoop 生态系统里一直占据独特的地位。

人们常说，打败一个事物的往往不是跟它类似的事物，而是一个全新的事物。

近几年来，随着云计算技术不断完善，众多云存储技术在成本上比HDFS 具有明显的优势。比如：AWS S3、Azure Blob Storage、谷歌云存储。1TB 的对象云存储没约只需要20美元左右，而HDFS 每月需要花100 美元左右。

正因为如此，谷歌的HDFS 服务只是将HDFS 操作转换成对象云存储，在API 层面上兼容HDFS 接口方式存储数据。

云计算存储技术除了在成本上具有巨大优势以外，对于HDFS 不适应的场景，在设计之初就做了充分的考虑，并重点优化。相对于HDFS 有着后发的技术优势。

总结

回到本节开头提到的那个问题，HDFS 到底是先驱还是先烈呢？

毫无疑问，HDFS 作为大数据存储技术的开创者，肯定是先驱。尽管HDFS 乃至整个Hadoop 大数据技术在日益完善的云技术面前并不占优势，但持续了十几年Hadoop 大数据技术在全球范围内有着众多用户。

这些用户已经为Hadoop 投入了巨大的人力和财力。目前来看Hadoop 技术生命力还会保持较长的一段时间。

关于HDFS 和Hadoop 未来的发展，你们怎么看呢？

14 Spark 为什么这么快

当你做成功一件事，千万不要等待着享受荣誉，应该再做那些需要的事。
—— 巴斯德

上节我们讲到了HDFS 的架构、原理以及云计算对HDFS 的影响。这一小节我们来一起学习一下Spark 计算框架。带着问题学习往往效率会比较高。这一小节我们将回答这个问题：Spark 为什么这么快？

Spark到底有多快

spark官网旁边有这样的一段文字：

Run workloads 100x faster.

Apache Spark achieves high performance for both batch and streaming data, using a state-of-the-art DAG scheduler, a query optimizer, and a physical execution engine.

作为Hadoop MapReduce 后继者Apache Spark 可以支撑数千节点规模的集群部署，尤其在内存数据处理上，Spark比MapReduce更加高效，且支持GB或TB级别的数据。

然而很多人都认为在磁盘数据计算上，MapReduce 比Spark 更有优势。2014年，大数据公司Databricks 为了评估Spark 在PB 级磁盘数据计算的运行状况, 其技术团队使用 AWS进行了一个Daytona Gray 类别的排序基准测试。

测试结果显示Spark 打破了MapReduce 保持的排序性能记录。这次测试是一个考量系统排序100TB数据（约万亿条记录）速度的行业基准测试。

在此之前，这项基准测试的世界记录保持者是雅虎，他们使用2100 节点的MapReduce 集群在72 分钟内完成了计算。而本次测试Spark 只使用了206 个EC2 节点，就将排序用时缩短到了23 分钟。

也就是说在相同数据的排序上，Spark 只使用了1/10 的计算资源就比MapReduce 快了近3 倍。

Spark为什么这么快

Spark采用一种新的计算模型RDD。

RDD（Resilient Distributed Dataset）叫做弹性分布式数据集，是Spark 中最基本的数据抽象，它代表一个不可变、可分区、里面的元素可并行计算的集合。

Resilient：RDD 中的数据可以存储在内存中或者磁盘中。

Distributed：RDD 中的数据是分布式存储的，可用于分布式计算。

Dataset：一个数据集合，用于存放数据的。

RDD 具有数据流模型的特点：自动容错、位置感知性调度和可伸缩性。RDD 允许用户在执行多个查询时显式地将数据缓存在内存中，后续的查询能够重用这些数据，这极大地提升了查询速度。

RDD 之间的依赖关系分类窄依赖（narrow dependency）和宽依赖（wide dependency, 也称 shuffle dependency)。

窄依赖：是指每个父RDD 的一个Partition 最多被子RDD 的一个Partition 所使用，例如map、filter、union 等操作都会产生窄依赖，类比独生子女。

宽依赖：是指一个父RDD 的Partition 会被多个子RDD 的Partition 所使用，例如groupByKey、reduceByKey、sortByKey 等操作都会产生宽依赖。类比超生。

相比于宽依赖，窄依赖对优化很有利，主要基于以下两点：

宽依赖往往对应着shuffle 操作，需要在运行过程中将同一个父RDD 的分区传入到不同的子RDD 分区中，中间可能涉及多个节点之间的数据传输；而窄依赖的每个父RDD 的分区只会传入到一个子RDD 分区中，通常可以在一个节点内完成转换。

当RDD 分区丢失时（某个节点故障），Spark 会对数据进行重算。

对于窄依赖，由于父RDD 的一个分区只对应一个子RDD 分区，这样只需要重算和子RDD 分区对应的父RDD 分区即可，所以这个重算对数据的利用率是100% 的；
对于宽依赖，重算的父RDD 分区对应多个子RDD 分区，这样实际上父RDD 中只有一部分的数据是被用于恢复这个丢失的子RDD 分区的，另一部分对应子RDD 的其它未丢失分区，这就造成了多余的计算；

更一般的，宽依赖中子RDD 分区通常来自多个父RDD 分区，极端情况下，所有的父RDD 分区都要进行重新计算。

如下图所示如果子RDD2 的2A 分区计算失败，上游的父RDD1 的1A、1B、1C 三个分区都需要重新计算。会产生计算冗余，1A、1B、1C 三个分区对应的2B 数据也重新计算了。

上面我们建立了Spark RDD 的概念，以及RDD之间的依赖关系。那么整个Spark 程序是怎样组织RDD之间的协作关系呢？

Spark 程序会使用一种叫DAG（Directed Acyclic Graph）有向无环图的方式组织RDD逻辑关系。

在Spark程序中每个partition 分区是独立计算的最小个体。当RDD之间的依赖关系是窄依赖的时候，子RDD的某个只需要等依赖上游父RDD对应的某个partion分区计算完毕之后即可开始计算下游计算逻辑，而不必等上游父RDD全部partition 计算完毕。

所以Spark 根据宽依赖将整个DAG 图划分成多个Stage ,每个Stage执行完毕之后会记录一个CheckPoint 。当某个Stage 计算失败的时候，只需要恢复到上个CheckPoint 状态就可以了。

下面是Spark执行过程：

各个RDD之间存在着依赖关系，这些依赖关系就形成有向无环图DAG ，DAGScheduler 对这些依赖关系形成的DAG进行Stage划分，划分的规则很简单，从后往前回溯，遇到窄依赖加入本Stage ，遇见宽依赖进行Stage切分。

完成了Stage 的划分。DAGScheduler 基于每个Stage 生成TaskSet ,并将TaskSet 提交给TaskScheduler。TaskScheduler 负责具体的task调度,最后在Worker节点上启动task。

DAGScheduler

（1）DAGScheduler 对DAG 有向无环图进行Stage划分。

（2）记录哪个RDD 或者 Stage 输出被物化（缓存），通常在一个复杂的shuffle 之后，通常物化一下(cache 、persist )，方便之后的计算。

（3）重新提交shuffle 输出丢失的stage（stage内部计算出错）给TaskScheduler。

（4）将 Taskset 传给底层调度器

a）– spark-cluster TaskScheduler

b）– yarn-cluster YarnClusterScheduler

c）– yarn-client YarnClientClusterScheduler

TaskScheduler

（1）为每一个TaskSet 构建一个TaskSetManager 实例管理这个TaskSet 的生命周期。

（2）数据本地性决定每个task 最佳位置。

（3）提交 taskset ( 一组task ) 到集群运行并监控。

（4）推测执行，碰到计算缓慢任务需要放到别的节点上重试。

（5）重新提交Shuffle输出丢失的Stage 给DAGScheduler。

总结

我们来总结一下Spark计算速度快的原因：

1.高效DAG 有向无环图将复杂的计算逻抽象成一个个RDD ，并按照RDD之间的依赖关系，划分成多个Stage ，最大化地减少了重复计算。

2.RDD 丰富的计算算子可以匹配各种的计算场景，而不像MapReduce 需要将所有计算都转化成Map 、Reduce 两种计算模型。Spark RDD 的各个算子默认包含了大量的算法优化，RDD 计算效率对开发人员性能优化经验的依赖大大降低。

3.Spark RDD 之间转过过程中会优先使用内存缓存数据，相比MapReduce 计算模型Shuffle 阶段一定要通过磁盘缓存数据的方式，执行效率要高效得多。

4.Spark通过CheckPoint 机制最大化地减少了偶然计算失败导致的重复计算。

关于Spark 计算速度快的原因，你有哪些看法呢？

15 Strom、Spark streamming 、Flink 怎么选

读书而不思考，等于吃饭而不消化。
——波尔克

上一小节中我们讲到了Spark相对于MapReduce在数据批处理性能上的有着巨大飞跃，这小节我们来讲讲实时计算引擎的技术选型。

实时计算引擎相对于批处理计算引擎领域来说可选择的技术方案丰富很多，主流方案的包含Flink、Spark streamming，较老的Strom，比较小众的 Kafka、Pulsar，还有Google的Beam，Intel 的Gearpump，IBM 的Edgent等等。

主流实时计算引擎总体对比

产品 Spark Streaming Storm Flink
架构 依赖spark生态，主从模式，每个Batch处理都依赖主（driver），可以理解为时间维度上的spark DAG（有向五环图）。 主从模式，且依赖zookeeper，处理过程中对主的依赖不大。 架构介于spark streaming和storm之间，主从结构与spark streaming相似，DataFlow Grpah与Storm相似，数据流可以被表示为一个有向图。 每个顶点是一个用户定义的运算，每向边表示数据的流动。
容错 WAL及RDD 血统机制 Records ACK机制 基于Chandy-Lamport distributed snapshots checkpoint机制
延迟 处理的是一个事件窗口内的所有事件。具有秒级高延迟。 处理的是每次传入的一个事件可实现亚秒级低延迟。 同Strom，单条事件处理可实现亚秒级低延迟
吞吐量 一般 一般 最高
数据处理保证 exactly once（实现采用Chandy-Lamport 算法，即marker-checkpoint ），可靠性较高。 at least once(实现采用record-level acknowledgments)，Trident可以支持storm 提供exactly once语义。可靠性一般。 exactly once，可靠性较高
成熟度与活跃度 已经发展了一段时间，相对稳定 已经发展了一段时间，相对稳定。 新兴项目，蓬勃发展中，社区活跃度一直在上升。
易用性 支持SQL Steaming，Batch和STREAMING采用统一编程框架。 不支持SQL Steaming 支持SQL Steaming，Batch和STREAMING采用统一编程框架。
适用场景 不是真正的实时流处理框架，有历史包袱，但是其使用起来很简单，且天然对接Spark生态栈中的其他组件，社区支持较大，吞吐量较大，部署也很简单，UI也做的更加智能。所以，如果对延迟要求不高的情况下，建议直接使用Spark Streaming 即可。 大体上同 Spark Streaming，但是具备亚秒级延迟，速度上可以和 Flink 媲美，比起 Spark Streaming，其对实时性支持更好一些。但是在各方面的性能表现上，比 Flink 较差。 完全为实时流处理而诞生，没有历史包袱。如果对延迟性要求比较高的话，那么建议直接上 Flink。除了延迟较低外，其在API和容错上也是做的比较完善，使用起来相对来说也是比较简单的，部署容易，而且发展势头也越来越好。
Storm

Storm是一种较早期的实时计算引擎。像MapReduce 是专门为数据批处理设计一样，Storm 从一开始设计的时候就是为实时计算而生的，一经推出一度成为实时计算的工业标准。但是Storm 受限于当时历史背景，很多方面的设计并不优雅。Storm 的Ack机制、exactly once 可靠性差。

在Storm 出现之前，进行实时处理是非常痛苦的事情，我们主要的时间都花在关注往哪里发消息，从哪里接收消息，消息如何序列化，真正的业务逻辑只占了源代码的一小部分。

Storm 出现之后，人们具备了处理实时计算的能力，但是Storm exactly once 可靠性差、Ack 开启之后性能差的问题，让Storm 的应用范围局限在Lambda 架构之上处理小数据量的实时计算问题。

来源Storm官网:http://storm.apache.org/

Storm 主要架构有Nimbus、ZooKeeper、Supervisor 组成。

Nimbus：负责在集群里面发送代码，分配工作给机器并且监控状态，在集群中只有一个，作用类似Hadoop 里面的JobTracker。

ZooKeeper：Storm 重点依赖的外部资源，Nimbus、Supervisor 和Worker 等都是把心跳数据保存在ZooKeeper上，Nimbus 也是根据ZooKeeper上的心跳和任务运行状况进行调度和任务分配的。

Supervisor：在运行节点上，监听分配的任务，根据需要启动或关闭工作进程Worker。每一个要运行Storm 的机器上都运行一个Supervisor，并且按照机器的配置，设定上面分配的槽位数。

Topology：Storm 提交运行的程序称为Topology，它是由Spout 和Bolt 构成，Spout 是发出Tuple 的结点，Bolt可以随意订阅某个Spout 或者Bolt 发出的Tuple。

Spout：Spout 是一个Topology 的消息生产的源头，Spout 生产的消息在Storm 中被抽象为Tuple，在整个Topology 的多个计算组件之间都是根据需要抽象构建的Tuple 消息来进行连接，从而形成流。实时数据源的处理逻辑在spout 里配置。

Bolt：Storm中消息的处理逻辑被封装到Bolt组件中，任何处理逻辑都可以在Bolt 里面执行，实时数据处理的业务逻辑写在Bolt里。

下图的表格列举了Storm 和Hadoop 概念的对应关系，熟悉Hadoop 的同学应该能够很好的理解了。

Storm Hadoop
Nimbus JobTracker
Supervisor TaskTracker
Topology Job
Spark Streamming

Spark Streamming处理实时数据的思路跟Strom 有很大的不同，Spark Streamming 将无限实时数据集分割成小的数据集，基于这些小的数据集，进行微批处理计算。

Spark Streaming 有高吞吐量和容错能力强这两个特点。Spark Streaming 支持的数据输入源很多，例如：Kafka、Flume、Twitter、ZeroMQ 和简单的 TCP 套接字等等。

数据输入后可以用 Spark 的高度抽象原语如：map、reduce、join、window 等进行运算。而结果也能保存在很多地方，如 HDFS，数据库等。另外 Spark Streaming 也能和 MLlib（机器学习）以及 Graphx 完美融合。

实时无限数据集输入以后会分割成多个批处理的数据，批处理的数据然后经过一系列的加工逻辑，然后会被输出成流处理的结果。

Spark Streaming 提供了一种叫DStream 的 API 处理实时数据流。DStream 将Spark Streaming 处理数据的底层实现的细节做了高度抽象，用户只需要了解DStream 的算子即可处理实时无限数据集。

DStream 目前有微批和流式两个处理引擎。微批提供了窗口、消息时间、trigger、watermarker、流表 join、流流 join 等常见的流处理能力，时延仍然保持最小 100 毫秒。

DStream 的流式引擎还处在实验阶段，虽然延迟上做到了1毫秒，不支持 exactly-once 语义，只支持 at-least-once 语义。

DStream 底层的数据处理操作都是基于RDD 实现的，Spark RDD 身上具备的高容错性、数据处理高效性优点都被DStream 继承了。

上面的图展示了一个DStream 的flatMap 算子的实现原理。DSteam 的数据流按照时间进行分割成多个小的数据集，每个数据集通过flatMap 算子转换成新的数据集，这些新的数据集也是基于时间排序，最终转换成一个新的DStream。

Flink

Flink 项目由德国柏林理工大学教授 Volker Markl 发起，他参考了谷歌的流计算最新论文 MillWheel，决定以流计算为基础，开发一个流批结合的分布式流计算引擎 Flink。Flink 于 2014 年 3 月进入 Apache 孵化器并于 2014 年 11 月毕业成为 Apache 顶级项目。

Flink 处理数据的设计思路是流批合一，以流为基础，批是流的特例或上层 API。

Flink 采用 Dataflow 模型，和 Lambda 模式不同。Dataflow 的逻辑关系和运行时的物理拓扑相差不大，是纯粹的流式设计，时延和吞吐理论上是最优的。

Flink 在流批计算上没有包袱，一开始就走在对的路上。

Flink 的架构上由JobManagers、TaskManagers、Clients组成。

当用户提交一个 Flink 程序时，会首先创建一个 Client，该 Client 首先会对用户提交的 Flink 程序进行预处理，并提交到 Flink 集群中处理。

Client 会将用户提交的Flink程序组装一个JobGraph， 并且是以JobGraph 的形式提交的。一个JobGraph 是一个Flink Dataflow，它由多个JobVertex 组成的DAG。其中，一个JobGraph 包含了一个Flink 程序的如下信息：JobID、Job名称、配置信息、一组JobVertex等。

JobManagers （也称为 masters）协调分布式计算。它们负责调度任务、协调 checkpoints、协调故障恢复等。

每个 Job 至少会有一个 JobManager。高可用部署下会有多个 JobManagers，其中一个作为 leader，其余处于 standby 状态。

TaskManagers（也称为 workers）执行 dataflow 中的 tasks（准确来说是 subtasks ），并且缓存和交换数据 streams。

每个 TaskManager（worker）都是一个 JVM 进程，并且可以在不同的线程中执行一个或多个 subtasks。为了控制 worker 接收 task 的数量，worker 拥有所谓的 task slots （至少一个）。

每个 task slots 代表 TaskManager 的一份固定资源子集。

通过调整 slot 的数量，用户可以决定 subtasks 的隔离方式。每个 TaskManager 有一个 slot 意味着每组 task 在一个单独的 JVM 中运行（例如，在一个单独的容器中启动）。

拥有多个 slots 意味着多个 subtasks 共享同一个 JVM。 Tasks 在同一个 JVM 中共享 TCP 连接（通过多路复用技术）和心跳信息（heartbeat messages）。它们还可能共享数据集和数据结构，从而降低每个 task 的开销。

默认情况下，Flink 允许 subtasks 共享 slots，即使它们是不同 tasks 的 subtasks，只要它们来自同一个 job。

通过 slot sharing，将示例中的并行度从 2 增加到 6 可以充分利用 slot 的资源，同时确保繁重的 subtask 在 TaskManagers 之间公平地获取资源。

总结
Flink 基于流的流批一体的思路，相对于Spark Streaming 基于批的批流一体设计，在设计思路上就领先一大步，直接的收益就是Flink 处理数据延迟在亚秒级，小于Spark Streaming 秒级的延迟。同时在数据处理吞吐量上面相对于Storm 有着巨大的优势，同时由于Flink高可靠的exactly once设计，Flink 在数据精确去重场景下对于Storm 有着明显的优势。

16 Hbase：列式存储利器

人要有毅力，否则将一事无成。
——居里夫人

上一节我们一起学习实时计算引擎的技术选型，这一节来学习Hbase 数据库。

大家都知道Google 通过GFS、MapReduce、Big Table "三驾马车"开创了大数据技术的先河。这三篇论文像一座灯塔照亮了开源大数据技术行业的发展。

基于GFS 的设计，产生了HDFS 分布式文件系统，用于大数据的文件存储。基于MapReduce，产生了Java 版本的MapReduce，用做大数据的计算引擎。而基于Big table，产生了Hbase 分布式数据库，用于大数据的查询。
图片描述

HBase是什么

HBase 是一个高可靠、高性能、面向列、可伸缩的分布式NoSQL 数据库，运行于HDFS 文件系统之上，主要用来存储非结构化和半结构化的松散数据。

HBase 的目标是处理非常庞大的表，可以通过水平扩展的方式，利用廉价计算机集群处理由超过10 亿行数据和数百万列元素组成的数据表。

HBase VS HDFS

HDFS 本质上是一个数据文件系统，对于数据批量查询的场景有着很好的支持，但对于数据随机查询上比较无力。同时HDFS不支持数据的更新、不合适增量数据处理。为了解决大规模数据的随机查询问题，HBase 应运而生。

HBase VS RDBMS

一直以来人们都在使用传统的关系型数据库（RDBMS）解决数据写入、数据查询问题。关系型数据库在解决数据查询上的表现是久经考验的，具有轻量化、高可靠性、索引支持良好的优点，并支持事务特性。

但是在大数据场景下，RDBMS 查询效率非常低。例如RDBMS 里面应用广泛的Mysql 数据库，单个数据表的数据量超过千万级以后数据查询效率会直线下降。然而在大数据场景下，单个数据表数据量过亿的场景都非常常见。

Hbase RDBMS
数据库大小 PB GB、TB
数据类型 Bytes 丰富的数据类型
事务支持 ACID只支持单个Row级别 全面的ACID支持ROW和表
索引 只支持ROW-Key 支持
吞吐量 百万查询/秒 数万查询/秒
正因为RDBMS 不能满足大数据场景下数据查询性能要求，数据库科学家们一直在尝试探索新的数据库系统。直到人们突破关系型数据库思维束缚，创造出非关系型数据库，也就是Nosql 数据库之后，一切问题迎刃而解。

Hbase 正是基于Nosql 的设计思想，并添加列式存储、高效的数据模型设计，针对大数据场景做了优化，一经推出以后迅速占领了工业级大数据查询领域的市场。

由于Hbase 的巨大影响力，一个名叫Phoenix 项目实现在Hbase 之上使用SQL 语法查询数据。Phoenix 最终进入Apache 基金会，成为了顶级的开源项目。

借助Phoenix 对于SQL查询语法的支持，Hbase 的数据查询使用门槛被进一步降低。

Hbase架构

Hbase 是基于Zookeeper、HDFS 构建。Hbase自身的主要组件有：HMaster、Region Server。
图片描述

Zookeeper：

Zookeeper 是HBase HA 的解决方案，是整个集群的协调器通过Zookeeper 保证了至少有一个HMaster 处于active 状态HMaster 并不直接参与数据的读写操作。

当我们使用HBase 的API 的时候，当我们想用HBase 的API 去读取数据的时候，我们并不需要知道HMaster 的地址、也不需要知道RegionServer 的地址，我们只需要知道Zookeeper 集群的地址就可以了

HMaster 启动将系统加载到Zookeeper

Zookeeper 保存了HBase 集群region 的信息、meta 的信息等等

维护着RegionServer 的状态信息，知道哪些数据需要从哪些RegionServer 去读取

HMaster：

HMaster 是HBase 主/从集群架构中的中央节点
HMaster 用于协调多个RegionServer、检测各个RegionServer 的状态、并且平衡各个RegionServer 之间的负载、同时还负责分配region 到RegionServer
region：region 是HBase 中存储的最小的单元、是HBase 表格的基本单位
HMaster 维护表和Region 的元数据，不参与数据的输入/输出过程
HBase 本身是支持HA 的，也就是说同时可以有多个HMaster 进行运行，但是只有1个处于active 状态
如果处于active 的节点失效了，挂掉了，其它的HMaster 节点就会选举出一个active 节点来接管整个HBase 集群
Region Server：

Region Server 负责数据的读写，数据存放在内存中，持续化需要和HDFS 文件系统进行I/O 交互。

HBase 是列族数据库，列的数据是存放在一起的，不同的行按照row key 分布，存储在不同的Region Server中。

HDFS:

数据的存储与备份。

将数据存储在HDFS 的一个显而易见的好处时，当集群Region Server 发生变化时，增加或者减少时，不需要在节点间进行数据的复制，这大大减少了节点的上下线时间，和I/O 消耗。

Hbase数据模型

在Hbase中数据被存在Table 中的，但是Hbase 的Table 跟RDBMS 里面的Table 有很大的不同，Hbase 的Table 是一个多维的Map。Hbase 数据模型的逻辑视图是由下面概念组成的。

Row：在表里面，每一行代表着一个数据对象，每一行都是以一个Row Key 来进行唯一标识的，Row Key 并没有什么特定的数据类型，以二进制的字节来存储

Column：HBase 的列由Column family 和Column qualifier 组成，由冒号（:）进行间隔。

比如 family:qualifier

RowKey：可以唯一标识一行记录，不可被改变。改变的唯一方式是删除这个RowKey，再重新插入

Column Family：是一些Column 的集合，1个Column Family 所包含的所有的Column 成员是有着相同的前缀。在物理上1个Column Family 所有的成员是存储在一起的，存储的优化都是针对Column Family 级别的。

这就意味着1个- Column Family 的成员都是用相同的方式进行访问的。在定义HBase 表的时候需要提前设置好列族，表中所有的列都需要组织在列族里面；列族一旦定义好之后，就不能轻易的更改了，因为它会影响到HBase真实的物理存储结构。

Column Qualifier：列族中的数据通过列标识（Column Qualifier）来进行映射，可以理解为一个键值对，Column Qualifier 就是key

Cell：每一个RowKey、Column Family、Column Qualifier 共同组成的一个单元。存储在Cell 里面就是我们想要保存的数据。Cell存储的数据没有特定的数据类型，以二进制字节来进行存储

Timestamp：每个值都会有一个timestamp，作为该值特定版本的标识符。默认HBase 中每次插入数据的时候，都会用timestamp 来进行版本标识。读取数据时，如果这个时间戳没有被指定，就默认返回最新的数据。

写入数据时，如果没有设置时间戳，默认使用当前的时间戳。每一个列族的数据的版本都由HBase 单独维护。默认情况下，HBase 会保留3 个版本的数据

Row Key Time Stamp ColumnFamily contents ColumnFamily anchor ColumnFamily people
“com.cnn.www” t9 anchor:cnnsi.com = “CNN”
“com.cnn.www” t8 anchor:my.look.ca = “CNN.com”
“com.cnn.www” t6 contents:html = “…”
“com.cnn.www” t5 contents:html = “…”
“com.cnn.www” t3 contents:html = “…”
“com.example.www” t5 contents:html = “…” people:author = “John Doe”
上面是一个实际Hbase 逻辑视图的例子。每一行有一个唯一的row key，每个Row key对应着三个Column Family，每个ColumnFamily 是有一个后者多个key-value 对组成的。这些key就是Column Qualifier，Value 则以Cell 保存的数据。逻辑视图表示成Map 的形式是下面这样的。

{
  "com.cnn.www": {
    contents: {
      t6: contents:html: "<html>..."
      t5: contents:html: "<html>..."
      t3: contents:html: "<html>..."
    }
    anchor: {
      t9: anchor:cnnsi.com = "CNN"
      t8: anchor:my.look.ca = "CNN.com"
    }
    people: {}
  }
  "com.example.www": {
    contents: {
      t5: contents:html: "<html>..."
    }
    anchor: {}
    people: {
      t5: people:author: "John Doe"
    }
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

虽然Hbase 逻辑视图上，多个ColumnFamily 和Row Key 组成一个Row（行），但是ColumnFamily 物理上是分开存储的。下面两张表是Hbase 数据模型实际的存储形式。

Row Key Time Stamp Column Family anchor
“com.cnn.www” t9 anchor:cnnsi.com = “CNN”
“com.cnn.www” t8 anchor:my.look.ca = “CNN.com”
Row Key Time Stamp ColumnFamily contents:
“com.cnn.www” t6 contents:html = “…”
“com.cnn.www” t5 contents:html = “…”
“com.cnn.www” t3 contents:html = “…”
Hbase 的Column Family 是包含在Table Schema 信息里面的，需要提前定义好。Column Qualifier 不是Schema信息里面的，可以随时添加新的Column Qualifier 到Hbase 的Table 当中。

总结

Hbase 是基于Nosql 的Key-Value 数据模型，独特的列式存储方式对于大数据场景下的数据查询有着极高的效率，同时宽松的ColumnFamily 设计，摆脱了传统RDBMS 严格的Scheme 束缚，更好的适应了大数据场景下数据模型的快速更新需求。

Phoenix 项目的出现让开发人员可以使用SQL 语言的方式查询Hbase 的数据，让Hbase 成了Hadoop 大数据技术最重要的组成部分。

17 Yarn：资源调度中枢

什么是路？就是从没路的地方践踏出来的，从只有荆棘的地方开辟出来的。 
—— 鲁迅

上一小节我们学习了Hbase 数据库，这一小节我们来一起学习Yarn。

YARN的发展历史

Hadoop 在框架演进的过程中经历了多个阶段，其中主要包括四个阶段：Ad Hoc (点对点)集群时代(单用户)、Hadoop on Demand(HOD) 时代(以通用系统形式，在共享集群中部署私有Hadoop MapReduce 和HDFS 实例)、共享集群黎明时代( JobTracker 和 TaskTracker 的诞生)、YARN 时代。

在这个过程中Hadoop 碰到了需要需要解决的问题，而YARN 就是为了解决这些具体问题而设计的。

YARN 需要解决的问题：

1、可扩展性：下一代计算平台应该可以平滑地扩展到数万个节点和并发的应用。

2、可维护性：下一代计算平台应该保证集群升级与用户应用程序的完全解耦。

3、多租户：下一代计算平台需要支持一个集群中多个租户并存，同时支持多个租户之间细粒度地共享单个节点。

4、位置感知：对很多应用来说，将计算移动到数据所在的位置是一个重大的进步。

5、高集群使用率：下一代计算平台底层物理资源的高使用率。

6、安全和可审计操作：下一代计算平台继续以安全的、可审计的方式使用集群资源

7、可靠性和可用性：下一代计算平台应该有高度的可靠的用户交互，并支持高可用性。

8、对编程模型多样性的支持：下一代计算平台必须支持多样化的编程模型，需要演进不仅仅以MapReduce 为核心。

9、灵活的资源模型：下一代计算平台支持各个节点的动态资源配置以及灵活的资源模型。

10、向后兼容：下一代计算平台应该保持现有MapReduce 应用程序的完全向后兼容。

YARN 诞生之前计算资源的管理是在MapReduce 中实现的，JobTracker 和TaskTracker 承担了MapReduce 计算过程中计算资源的调度管理任务。YARN 诞生之后计算资源的管理功能被单独抽离出来。

YARN 从设计之初就将兼容编程模型多样化作为核心要点，不仅可以支持MapReduce On Yarn，还可以兼容其他后出现的计算引擎。

因为Yarn较为优秀的通用性，MapReduce 之后出现的Spark、Flink 都可以通过Yarn 管理计算资源，YARN 成为了应用最为广泛的通用资源调度系统。

如果你的应用程序需要使用YARN 的资源管理功能，也可以实现YARN 提供的API，来将应用运行在YARN 之上，将资源分配和回收统一交给YARN 去管理，大大简化资源管理功能的开发。

JobTracker和TaskTracker架构

JobTacker其承担的任务有：接受任务、计算资源、分配资源、与DataNode 进行交流。

在hadoop中每个应用程序被表示成一个作业，每个作业又被分成多个任务，JobTracker 的作业控制模块则负责作业的分解、TaskTracker 状态监控、作业状态监控、任务状态监控。

TaskTracker 是JobTracker 和Task 之间的桥梁。

一方面，从JobTracker 接收并执行各种命令：运行任务、提交任务、杀死任务等；另一方面，将本地节点上各个任务的状态通过心跳周期性汇报给JobTracker。TaskTracker 与JobTracker 和Task 之间采用了RPC 协议进行通信。

JobTracker 和TaskTracker 组成的资源调度架构缺点比较多：

JobTracker 是集群事务的集中处理点，存在单点故障

JobTracker 需要完成的任务太多，既要维护job 的状态又要维护job 的task 的状态，造成过多的资源消耗

在taskTracker 端，用map/reduce task 作为资源的表示过于简单，没有考虑到CPU、内存等资源情况，当把两个需要消耗大内存的task 调度到一起，很容易出现OOM

把资源强制划分为map/reduce slot ,当只有map task 时，reduce slot 不能用。当只有reduce task 时，map slot 不能用，容易造成资源利用不足。

YARN架构

Yarn的架构主要由：Resource Manager、Node Manager、Application Mater、Container组成。

YARN最基本的想法是将原JobTracker 主要的资源管理和job调度/监视功能分开作为两个单独的守护进程。

有一个全局的ResourceManager(RM )和每个Application 有一个ApplicationMaster(AM)，Application map-reduce job 或者DAG jobs。

ResourceManager 和 NodeManager(NM) 组成了基本的数据计算框架。ResourceManager协调集群的资源利用，任何client 或者运行着的applicatitonMaster 想要运行job 或者task 都得向RM 申请一定的资源。

RM使用resource container 概念来管理集群的资源，resource container 是资源的抽象，每个container 包括一定的内存、IO、网络等资源，不过目前的实现只包括内存一种资源。

ApplicationMaster

ApplicationMaster 首先是一个框架库，它的功能在官网上说的不够系统，大意，由于NodeManager 执行和监控任务需要资源，所以通过ApplicationMaster 与ResourceManager 沟通，获取资源。换句话说，ApplicationMaster 起着中间人的作用。

转换为更专业的术语：AM负责向ResourceManager索要NodeManager执行任务所需要的资源容器，更具体来讲是ApplicationMaster 负责从Scheduler 申请资源，以及跟踪这些资源的使用情况以及任务进度的监控。

对于MapReduce 框架而言有它自己的AM 实现，用户也可以实现自己的AM，在运行的时候，AM 会与NM 一起来启动和监视tasks。

NodeManager

NodeManager 是基本的计算框架。NodeManager 是客户端框架负责 containers, 监控他们的资源使用 (cpu, 内存, 磁盘, 网络) 和上报给 ResourceManager/Scheduler。

在启动container 的时候，NM会设置一些必要的环境变量以及将container 运行所需的jar 包、文件等从hdfs 下载到本地，也就是所谓的资源本地化。

当所有准备工作做好后，才会启动代表该container 的脚本将程序启动起来。启动起来后，NM会周期性的监视该container 运行占用的资源情况，若是超过了该container 所声明的资源量，则会kill掉该container 所代表的进程。

另外，NM 还提供了一个简单的服务以管理它所在机器的本地目录。Applications 可以继续访问本地目录即使那台机器上已经没有了属于它的container 在运行。例如，Map-Reduce 应用程序使用这个服务存储map output 并且shuffle 它们给相应的reduce task。

在NM 上还可以扩展自己的服务，YARN提供了一个yarn.nodemanager.aux-services 的配置项，通过该配置，用户可以自定义一些服务，例如Map-Reduce 的shuffle 功能就是采用这种方式实现的。

ResourceManager

ResourceManager 有两个组件：调度器(Scheduler)和应用程序管理器(ApplicationsManage)。

调度器（Scheduler）是可插拔的，比如有Fair Scheduler、Capacity Scheduler等，当然调度器也可以自定义。

Scheduler 负责分配最少但满足application 运行所需的资源量给Application。

Scheduler 只是基于资源的使用情况进行调度，并不负责监视/跟踪application 的状态，当然也不会处理失败的task。

ApplicationsManager 负责处理client 提交的job 以及协商第一个container 以供applicationMaster 运行，并且在applicationMaster 失败的时候会重新启动applicationMaster。

ApplicationsManager 负责系统中所有AM的生命周期的管理。ApplicationsManager 负责AM ，当AM 启动后，AM会周期性的向ApplicationsManager 发送heartbeat，默认是1s，AsM 据此了解AM 的存活情况，并且在AM 失败时负责重启AM。

若是一定时间过后(默认10分钟)没有收到AM 的heartbeat，AsM 就认为该AM 失败了。

Container

Container 是YARN 的资源抽象，它封装了某个节点中的内存、CPU、磁盘、网络等，当ApplicationMaster 向ResourceManager申请资源的时候，RM向AM分配的资源就是通过Container 表示的。

YARN 会为每个任务分配一个Container，并且该任务只能使用自己Container 中的资源。Container 是一个动态的资源划分，根据应用程序的需求动态生成的。它不同于MRv1中的slot。

目前为止YARN 仅支持CPU 和内存两种资源，并且使用的是Cgroups 轻量级资源隔离机制进行隔离。

Yarn调度器与队列

在YARN中，调度器是一个可插拔的组件，常见的有FIFO，CapacityScheduler，FairScheduler。可以通过配置文件选择不同的调度器。

在RM端，根据不同的调度器，所有的资源被分成一个或多个队列（queue），每个队列包含一定量的资源。用户的每个application，会被唯一的分配到一个队列中去执行。

队列决定了用户能使用的资源上限。所谓资源调度，就是决定将资源分配给哪个队列、哪个application 的过程。

可见调度器的两个主要功能：1.决定如何划分队列；2.决定如何分配资源。此外，还有些其他的特性：ACL、抢占、延迟调度等等

总结

在YARN 的架构设计中，JobTracker 的功能被分散到各个进程中包括ResourceManager 和NodeManager：
比如监控功能，分给了NodeManager，和Application Master。

ResourceManager 里面又分为了两个组件：调度器及应用程序管理器。也就是说YARN 重构后，JobTracker 的功能，被分散到了各个进程中。同时由于这些进程可以被单独部署所以这样就大大减轻了单点故障，及压力。

18 Kafka：数据传输管道

我们有力的道德就是通过奋斗取得物质上的成功；这种道德既适用于国家，也适用于个人。
——罗素

这一节我们一起来学习一下数据传输管道：Kafka。

人们每天的日常生活会产生大量的数据，这些数据散落在各个角落，我们需要一套高效的数据传输工具来将这些数据集中起来。数据从产生到被集中收集的过程，就像下雨的时候雨水从城市街道里面的下水管道流入污水处理中心的过程一样。

由于数据量庞大，数据传输过程中既要保证不丢数据，又要保证下游数据读取数据时，数据能够被快速读取，数据传输工具需要设计的足够高效。另一方面。数据的产生速度是不稳定的，数据的产生量存在波峰和波谷，数据传输工具需要有”削峰填谷“的能力。

基于上面的需求而设计的数据传输工具通常被称之为消息系统。

点对点 VS 发布/订阅

消息系统根据消息被消费的特点分为：点对点模式(Peer-to-Peer)、发布/订阅模式。

点对点模式

消息生产者生产消息发送到queue 中，然后消息消费者从queue 中取出并且消费消息。这里要注意：

消息被消费以后，queue 中不再有存储，所以消息消费者不可能消费到已经被消费的消息。
Queue支持存在多个消费者，但是对一个消息而言，只会有一个消费者可以消费。
发布/订阅模式

消息生产者（发布）将消息发布到topic 中，同时有多个消息消费者（订阅）消费该消息。和点对点方式不同，发布到topic 的消息会被所有订阅者消费。

上面图中上半部分是一个点对点消息系统，Producer1 产生的消息放入消息系统中，Consumer1 消费了这个消息，其他的Consumer 不再能消费到这个消息了。上图中的下半部分是一个发布/订阅消息提供。Pushlisher1 发布的消息放入到消息系统之后，下游多个订阅者(Subscriber1~SubscriberN)都可以消费到这个消息。

发布/订阅模式的消息系统相对于点对点模式的消息系统，在消息消费模式上更加灵活，可以适应更加复杂的数据传输场景。

在大数据场景中，数据的生产和消费关系非常复杂，基于发布/订阅模式的消息系统被广泛运用于工业级大数据生产中。

Why Kafka

开源的消息系统非常多，有ActiveMQ、RabbitMQ、Kafka、RocketMQ。主要从以下几个维度进行对比：社区活跃度、客户端支持、吞吐量、延迟、数据可靠性。

数据可靠性 延迟 单机吞吐 社区 客户端
ActiveMQ 中 / 万级 不太活跃 支持全面
RabbitMQ 高 微秒级 万级 活跃 支持全面
Kafka 高 毫秒级 十万 活跃 支持全面
RocketMQ 高 毫秒级 十万 有待加强 待加强
Kafka 通过Consumer Group 的方式实现了同时支持点对点模型和发布/订阅模式传递消息。

Kafka背景

kafka 是最初由Linkedin 公司开发，使用Scala 语言编写，Kafka 是一个分布式、分区的、多副本的、多订阅者的日志系统(分布式MQ系统)，可以用于web/nginx 日志，搜索日志，监控日志，访问日志等等。

kafka 目前支持多种客户端语言：java，python，c++，php等等。

Kafka架构

Kafka 架构主要由：Producer、Broker、Consumer、Zookeeper 组成。

Broker：Kafka 集群包含一个或多个服务器，这种服务器被称为broker。

Producer：负责发布消息到Kafka broker。

Topic：每条发布到Kafka集群的消息都有一个类别，这个类别被称为topic。（物理上不同topic 的消息分开存储，逻辑上一个topic 的消息虽然保存于一个或多个broker上但用户只需指定消息的topic 即可生产或消费数据而不必关心数据存于何处。

Partition：parition 是物理上的概念，每个topic 包含一个或多个partition，创建topic 时可指定parition 数量。每个partition 对应于一个文件夹，该文件夹下存储该partition 的数据和索引文件。

Consumer：消费消息。每个consumer 属于一个特定的consumer group（可为每个consumer指定group name，若不指定group name 则属于默认的group ）。使用consumer high level API 时，同一topic 的一条消息只能被同一个consumer group内的一个consumer 消费，但多个consumer group 可同时消费这一消息。

Consumer Group （CG）：若干个Consumer 组成的集合。这是kafka 用来实现一个topic 消息的广播（发给所有的consumer）和单播（发给任意一个consumer）的手段。一个topic 可以有多个CG。

topic 的消息会复制（不是真的复制，是概念上的）到所有的CG，但每个CG 只会把消息发给该CG 中的一个consumer 。如果需要实现广播，只要每个consumer 有一个独立的CG 就可以了。

要实现单播只要所有的consumer 在同一个CG。用CG 还可以将consumer 进行自由的分组而不需要多次发送消息到不同的topic。

Kafka设计上的亮点

kafka 之所以被广泛运用到大数据工业级生产环境跟它本身非常多的设计亮点是分不开的。

高可用（High available）

Kafka通过replica 和isr 机制来保证数据的高可用。要了解这两个机制的原理，我们需要先了解一下Kafka 数据模型分层设计。

Kafka的数据模型上分为Topic（主题）-Partition（分区）-Message（消息）三层。

一个Topic对应多个Partition，replica（多副本机制）是对Patition 而言的，每个Partition 的副本个数N是可以配置的。对于一个Topic 的某个Patition 来说，会有一个replica 为leader，其他都为follower，leader 处理对这个partition 的所有读写请求，与此同时，follower 会被动地去复制leader 上的数据。

Kafka 提供日志复制算法保证,如果leader 发生故障或挂掉，一个新leader 被选举并且客户端的消息成功写入。Kafka 确保从isr (同步副本列表)中选举一个副本为leader ,或者换句话说,follower 追赶leader 日志。leader 负责跟踪isr (同步副本列表)中所有follower 滞后状态。

当生产者发送一条消息到Broker ,leader 写入消息并复制到所有follower。消息提交之后才被成功复制到所有的同步副本。消息复制延迟受最慢的follower限制,重要的是快速检测慢副本,如果follower "落后"太多或者失效,leader 将会把它从isr(同步副本列表)中移除。

Kafka 分区的作用主要是提供负载均衡的能力，即实现系统的高伸缩性。

Kafka 的replica（副本机制）和isr（同步副本列表）则保障了高可用性。

高性能(High performance)

Kafka 单机的吞吐量可以达到十万级，通过sendfile 和pagecache 来实现zero copy 机制，顺序读写的特性使得用普通磁盘就可以做到很大的吞吐，相对来说性价比比较高。

pagecache（页缓存）是操作系统实现的一种主要的磁盘缓存，用来减少对磁盘I/O 的操作。具体就是把磁盘中的数据缓存到内存中，把对磁盘的访问变成对内存的访问。

所谓的zero copy(零拷贝)是指将数据直接从磁盘文件复制到网卡设备中，而不需要经由应用程序之手。

Kafka在读取文件的时候大量时候pagecache（页缓存），减少了对磁盘的读取，从内存读取数据，另一方面使用zero copy(零拷贝)技术将内存中读取到的数据直接复制到网卡设备中，进一步提高了数据读取的性能。

Linux 操作系统而言，零拷贝技术依赖于底层的sendfile() 方法实现。对应于Java 语言，FileChannal.transferTo()的底层实现就是sendfile()。

容错（fault tolerance）

Kafka 的容错主要通过controller 和coordinator 实现的。

controller 主要是做集群的管理。coordinator 主要做业务级别的管理。这两种角色都由Kafka里面的某个broker来担任，这样failover 就很简单，只需要选一个broker 来替代即可。

从这个角度来说Kafka 有一个去中心化的设计思想在里面， 但controller 本身也是一个瓶颈，可以类比于hadoop的namenode 。

总结

消息系统有点对点模式(Peer-to-Peer)、发布/订阅模式，Kafka 通过Consumer Group 的方式实现了对这两种消息传递方式的支持。由于Kafka 诸多优秀的设计，使得kafka 具备高可用（High available）、高性能(High performance) 以及足够的容错（fault tolerance）能力。

然而kafka 也并非完美，在Kafka 集群上的主题(Topic)很多的时候，Kafka 集群的性能就会大幅度下降，正是由于这样的问题，一个名叫Pulsar 的新一代消息系统开始逐渐发展起来。

19 Flume：数据采集的爪子

成功＝艰苦的劳动＋正确的方法＋少谈空话。
——爱因斯坦

上一小节中我们介绍了数据传输中需要使用消息系统来缓存数据，实现” 削峰填谷 “。这一小节我们来学习一下数据采集的爪子：Apache Flume。

我们为什么需要 Flume

我们有了消息系统以后，数据可以缓存到消息系统里面，供下游消费。但是在实际的生产场景中，需要搜集的数据通常会有两种形态出现。一种是以结构化数据的形式存储在服务器的数据库中，比如 Mysql 数据库。另外一种形态是以后台服务器的日志的形式出现的。

这些散落在服务器的日志，通常以日志文件的形式存储一个在固定的日志文件夹中。对于一个超大型互联网公司来说，服务器的数量非常庞大，可以达到惊人的万台以上。这么多的服务器日志文件怎么才能搜集到消息系统当中呢？

这时候就需要一个分布式的日志搜集工具。

Apache Flume 是一个分布式、高可靠、高可用的用来收集、聚合、转移不同来源的大量日志数据到中央数据仓库的工具。

Apache Flume 是 Apache 软件基金会（ASF）的顶级项目。

Flume 架构

参照下图可以看得出 Agent 就是 Flume 的一个部署实例， 一个完整的 Agent 中包含了三个组件 Source、Channel 和 Sink，Source 是指数据的来源和方式，Channel 是一个数据的缓冲池，Sink 定义了数据输出的方式和目的地。

Event 是 Flume 定义的一个数据流传输的最小单元。Agent 就是一个 Flume 的实例，本质是一个 JVM 进程，该 JVM 进程控制 Event 数据流从外部日志生产者那里传输到目的地（或者是下一个 Agent）。

Source 消耗由外部（如 Web 服务器）传递给它的 Event 。外部以 Flume Source 识别的格式向 Flume 发送 Event 。例如，Avro Flume Source 可接收从 Avro 客户端（或其他 FlumeSink）接收 Avro Event。

用 Thrift Flume Source 也可以实现类似的流程，接收的 Event 数据可以是任何语言编写的，只要符合 Thrift 协议即可。

当 Source 接收 Event 时，它将其存储到一个或多个 channel。该 channel 是一个被动存储器，可以保持 Event 直到它被 Sink 消耗。『文件 channel』就是一个例子 - 它由本地文件系统支持。

sink 从 channel 中移除 Event 并将其放入外部存储库（如 HDFS，通过 Flume 的 HDFS Sink 实现）或将其转发到流中下一个 Flume Agent（下一跳）的 Flume Source。

Agent 中的 source 和 sink 与 channel 存取 Event 是异步的。

Flume 的 Source 负责消费外部传递给它的数据（比如 web 服务器的日志）。外部的数据生产方以 Flume Source 识别的格式向 Flume 发送 Event。

Flume 可以设置多级 Agent 连接的方式传输 Event 数据。也支持扇入和扇出的部署方式，类似于负载均衡方式或多点同时备份的方式

你可以根据自己的业务需求来任意组合传输日志的 Agent 实例，上面这张图就是一个扇入方式的 Flume 部署方式，前三个 Agent 的数据都汇总到一个 Agent4 上，最后由 Agent4 统一存储到 HDFS。

Flume Source

Source 的种类非常多，可以适应各种日志数据源需求。

Spooling Directory Source

Spooling Directory Source 用来搜集服务器日志文件夹里面的日志，是应用最为广泛的 Source。

这个 Source 允许你把要收集的文件放入磁盘上的某个指定目录。它会将监视这个目录中产生的新文件，并在新文件出现时从新文件中解析数据出来。数据解析逻辑是可配置的。

在新文件被完全读入 Channel 之后会重命名该文件以示完成（也可以配置成读完后立即删除）。

与 Exec Source 不同，Spooling Directory Source 是可靠的，即使 Flume 重新启动或被 kill，也不会丢失数据。同时作为这种可靠性的代价，指定目录中的文件必须是不可变的、唯一命名的。

Flume 会自动检测避免这种情况发生，如果发现问题，则会抛出异常：

如果文件在写入完成后又被再次写入新内容，Flume 将向其日志文件（这是指 Flume 自己 logs 目录下的日志文件）打印错误并停止处理。
如果在以后重新使用以前的文件名，Flume 将向其日志文件打印错误并停止处理。
为了避免上述问题，生成新文件的时候文件名加上时间戳是个不错的办法。

尽管有这个 Source 的可靠性保证，但是仍然存在这样的情况，某些下游故障发生时会出现重复 Event 的情况。这与其他 Flume 组件提供的保证是一致的。

a1.channels = ch-1
a1.sources = src-1
a1.sources.src-1.type = spooldir
a1.sources.src-1.channels = ch-1
a1.sources.src-1.spoolDir = /var/log/apache/flumeSpool
a1.sources.src-1.fileHeader = true
上面是一个 Spooling Directory Source 配置样例。Spooling Directory Source 监控了”/var/log/apache/flumeSpool“日志文件夹。日志文件夹新增的日志数据会被当做放到”ch-1“的 channel 中。

Avro Source

Avro Source 监听 Avro 端口接收从外部 Avro 客户端发送来的数据流。如果与上一层 Agent 的 Avro Sink 配合使用就组成了一个分层的拓扑结构。

配置范例：

a1.sources = r1
a1.channels = c1
a1.sources.r1.type = avro
a1.sources.r1.channels = c1
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 4141
Flume channel

channel 是在 Agent 上暂存 Event 的缓冲池。 Event 由 source 添加，由 sink 消费后删除。

Memory Channel

内存 channel 是把 Event 队列存储到内存上，队列的最大数量就是 capacity 的设定值。它非常适合对吞吐量有较高要求的场景，但也是有代价的，当发生故障的时候会丢失当时内存中的所有 Event。

配置范例：

a1.channels.c1.type = memory
a1.channels.c1.capacity = 10000
a1.channels.c1.transactionCapacity = 10000
a1.channels.c1.byteCapacityBufferPercentage = 20
a1.channels.c1.byteCapacity = 800000
Kafka Channel

将 Event 存储到 Kafka 集群（必须单独安装）。Kafka 提供了高可用性和复制机制，因此如果 Flume 实例或者 Kafka 的实例挂掉，能保证 Event 数据随时可用。 Kafka channel 可以用于多种场景：

与 source 和 sink 一起：给所有 Event 提供一个可靠、高可用的 channel。
与 source、interceptor 一起，但是没有 sink：可以把所有 Event 写入到 Kafka 的 topic 中，来给其他的应用使用。
与 sink 一起，但是没有 source：提供了一种低延迟、容错高的方式将 Event 发送的各种 Sink 上，比如：HDFS、HBase、Solr。
由于依赖于该版本附带的 Kafka 客户端，Flume1.8 需要 Kafka 0.9 或更高版本。 与之前的 Flume 版本相比，channel 的配置发生了一些变化。
配置参数组织如下：

通常与 channel 相关的配置值应用于 channel 配置级别，比如：a1.channel.k1.type =
与 Kafka 相关的配置值或 Channel 运行的以 “kafka.” 为前缀（这与 CommonClient Configs 类似），例如：a1.channels.k1.kafka.topic 和 a1.channels.k1.kafka.bootstrap.servers。 这与 hdfs sink 的运行方式没有什么不同
特定于生产者 / 消费者的属性以 kafka.producer 或 kafka.consumer 为前缀
可能的话，使用 Kafka 的参数名称，例如：bootstrap.servers 和 acks
当前 Flume 版本是向下兼容的，但是第二个表中列出了一些不推荐使用的属性，并且当它们出现在配置文件中时，会在启动时打印警告日志。
配置范例：

a1.channels.channel1.type = org.apache.flume.channel.kafka.KafkaChannel
a1.channels.channel1.kafka.bootstrap.servers = kafka-1:9092,kafka-2:9092,kafka-3:9092
a1.channels.channel1.kafka.topic = channel1
a1.channels.channel1.kafka.consumer.group.id = flume-consumer
File Channel

文件 channel 使用默认的用户主目录内的检查点和数据目录的路径（说的就是上面的 checkpointDir 参数的默认值）。 如果一个 Agent 中有多个活动的文件 channel 实例，而且都是用了默认的检查点文件， 则只有一个实例可以锁定目录并导致其他 channel 初始化失败。 因此，这时候有必要为所有已配置的 channel 显式配置不同的检查点文件目录，最好是在不同的磁盘上。 此外，由于文件 channel 将在每次提交后会同步到磁盘，因此将其与将 Event 一起批处理的 sink/source 耦合可能是必要的，以便在多个磁盘不可用于检查点和数据目录时提供良好的性能。

配置范例：

a1.channels = c1
a1.channels.c1.type = file
a1.channels.c1.checkpointDir = /mnt/flume/checkpoint
a1.channels.c1.dataDirs = /mnt/flume/data
Flume sink

Kafka Sink

这个 Sink 可以把数据发送到 Kafka topic 上。目的就是将 Flume 与 Kafka 集成，以便系统可以处理来自各种 Flume Source 的数据。目前支持 Kafka 0.9.x 发行版。

Flume1.8 不再支持 Kafka 0.9.x（不包括 0.9.x）以前的版本。

Kafka Sink 使用 Event header 中的 topic 和其他关键属性将 Event 发送到 Kafka。 如果 header 中存在 topic，则会将 Event 发送到该特定 topic，从而覆盖为 Sink 配置的 topic。 如果 header 中存在指定分区相关的参数，则 Kafka 将使用相关参数发送到指定分区。 header 中特定参数相同的 Event 将被发送到同一分区。 如果为空，则将 Event 会被发送到随机分区。

Kafka Sink 还提供了 key.deserializer（org.apache.kafka.common.serialization.StringSerializer） 和 value.deserializer（org.apache.kafka.common.serialization.ByteArraySerializer）的默认值，不建议修改这些参数。

a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.kafka.topic = mytopic
a1.sinks.k1.kafka.bootstrap.servers = localhost:9092
a1.sinks.k1.kafka.flumeBatchSize = 20
a1.sinks.k1.kafka.producer.acks = 1
a1.sinks.k1.kafka.producer.linger.ms = 1
a1.sinks.k1.kafka.producer.compression.type = snappy
上面给出 Kafka Sink 的配置示例。Kafka 生产者的属性都是以 kafka.producer 为前缀。

Kafka 生产者的属性不限于下面示例的几个。此外，可以在此处包含您的自定义属性，并通过作为方法参数传入的 Flume Context 对象在预处理器中访问它们。

HDFS Sink

这个 Sink 将 Event 写入 Hadoop 分布式文件系统（也就是 HDFS）。 目前支持创建文本和序列文件。 它支持两种文件类型的压缩。 可以根据写入的时间、文件大小或 Event 数量定期滚动文件（关闭当前文件并创建新文件）。 它还可以根据 Event 自带的时间戳或系统时间等属性对数据进行分区。

存储文件的 HDFS 目录路径可以使用格式转义符，会由 HDFS Sink 进行动态地替换，以生成用于存储 Event 的目录或文件名。 使用此 Sink 需要安装 hadoop， 以便 Flume 可以使用 Hadoop 的客户端与 HDFS 集群进行通信。

配置范例：

a1.channels = c1
a1.sinks = k1
a1.sinks.k1.type = hdfs
a1.sinks.k1.channel = c1
a1.sinks.k1.hdfs.path = /flume/events/%y-%m-%d/%H%M/%S
a1.sinks.k1.hdfs.filePrefix = events-
a1.sinks.k1.hdfs.round = true
a1.sinks.k1.hdfs.roundValue = 10
a1.sinks.k1.hdfs.roundUnit = minute
上面的例子中时间戳会向前一个整 10 分钟取整。比如，一个 Event 的 header 中带的时间戳是 11:54:34 AM, June 12, 2012，它会保存的 HDFS 路径就是 /flume/events/2012-06-12/1150/00。

总结

在数据搜集的时候通常需要处理两类数据：存储在关系型数据库中的结构化数据、散落在服务器日志文件夹中的日志数据。日志数据分散在数以万计的文件夹中，需要一个高效的日志数据搜集工具。

Apache Flume 是一个分布式、高可靠、高可用的日志数据搜集工具。通过配置 Source、Channel、Sink 的方式实现灵活的日志搜集。

20 Kylin：以空间换时间

学习要注意到细处，不是粗枝大叶的，这样可以逐步学习、摸索，找到客观规律。
—— 徐特立

大家好，这一节我们学习一下 OLAP 分析中，使用非常广泛的分析引擎：kylin。

OLAP VS OLTP

在学习 Kylin 之前我们先了解一下什么是 OLAP。OLAP 是 OnLine Analytical Processing 的简称。Online 一般指查询延迟在秒级或毫秒级，可以实现交互式查询。

OLAP 的查询一般需要 Scan 大量数据，大多时候只访问部分列，聚合的需求（Sum，Count，Max，Min 等）会多于明细的需求（查询原始的明细数据）。

OLAP 的典型查询一般像：现在各种应用在年末会发布的大数据分析和统计应用，比如 2017 豆瓣读书报告，2017 豆瓣读书榜单，网易云音乐 2017 听歌报告。

OLAP 在企业中的一个重要应用就是 BI 分析，比如 2017 年最畅销的手机品牌 Top5。哪类人群最喜欢小米或华为手机等等。

OLAP 的特点：

专门用来做决策支持
历史的，总结的，统一的数据比具体的，独立的数据更重要
侧重于查询
查询吞吐量和相应时间是关键性能指标
与 OLAP 相对的是 OLTP。OLTP 是 Online Transaction Processing 的简称。Transaction 是指形成一个逻辑单元，不可分割的一组读 & 写操作。

OLTP 的查询一般只会访问少量的记录，且大多时候都会利用索引。在线的面向终端用户直接使用的 Web 应用：金融，博客，评论，电商等系统的查询都是 OLTP 查询，比如最常见的基于主键的 CRUD 操作。

OLTP 的特点：

专门用来做日常的，基本的操作
任务由短的，原子的，隔离的事务组成
处理的数据量在 G 级别
重视一致性和可恢复性
事务的吞吐量是关键性能指标
最小化并发冲突
OLTP 需要解决数据的增、删、改、查的问题，OLAP 需要解决数据聚合的问题。

ROLAP VS MOLAP

OLAP 有多种实现方法，根据存储数据的方式不同可以分为 ROLAP、MOLAP。

ROLAP 主要通过数据引擎强大的计算能力，瞬间聚合数据得到 OLAP 结果。MOLAP 则是提前计算聚合好数据模型，查询的时候只需要返回已经聚合好的数据结果就行了。

ROLAP 需要强大的关系型 DB 引擎支撑，长期以来，由于传统关系型 DBMS 的数据处理能力有限，所以 ROLAP 模式受到很大的局限性。随着分布式、并行化技术成熟应用，MPP 引擎逐渐表现出强大的高吞吐低时延计算能力，号称亿级秒开的引擎不在少数，ROLAP 模式可以得到更好的延伸。

单从业务实际应用考虑，性能在千万量级关联查询现场计算秒开的情况下，已经可以覆盖很多应用场景，具备应用的可能性。

MOLAP 表示基于多维数据组织的 OLAP 实现（Multidimensional OLAP）。以多维数据组织方式为核心，也就是说，MOLAP 使用多维数组存储数据。

多维数据在存储中将形成 “立方块（Cube）“的结构，在 MOLAP 中对” 立方块 “的” 旋转”、“切块”、“切片” 是产生多维数据报表的主要技术。特点是将细节数据和聚合后的数据均保存在 cube 中，所以以空间换效率，查询时效率高，但生成 cube 时需要大量的时间和空间。

What is KyLin

在开源 MOLAP 方案中，Apache Kylin 应用最为广泛，社区活跃度最高的解决方案。

Kylin 的核心思想是预计算，利用空间换时间来加速查询模式固定的 OLAP 查询。

Kylin 的理论基础是 Cube 理论，每一种维度组合称之为 Cuboid，所有 Cuboid 的集合是 Cube。 其中由所有维度组成的 Cuboid 称为 Base Cuboid，图中 (A,B,C,D) 即为 Base Cuboid，所有的 Cuboid 都可以基于 Base Cuboid 计算出来。 在查询时，Kylin 会自动选择满足条件的最 “小” Cuboid，比如下面的 SQL 就会对应 Cuboid（A,B）:

select xx from table where A=xx group by B

Kylin 架构

Kylin 自身的组件只有两个：JobServer 和 QueryServer 。 Kylin 的 JobServer 主要负责将数据源（Hive,Kafka）的数据通过计算引擎（MapReduce，Spark）生成 Cube 存储到存储引擎（HBase）中。

QueryServer 主要负责 SQL 的解析，逻辑计划的生成和优化，向 HBase 的多个 Region 发起请求，并对多个 Region 的结果进行汇总，生成最终的结果集。

在架构设计上，Kylin 的数据源，构建 Cube 的计算引擎，存储引擎都是可插拔的。Kylin 的核心就是这套可插拔架构，Cube 数据模型和 Cuboid 的算法。

Kylin 基于对 Cube 进行预计算来满足低延迟的查询处理。Kylin 的数据源就是 Hive 上的表，以星型模式存在。Kylin 的离线计算过程会读取 Hive 中的原始数据，将事实表和维表 join 起来，然后对各种维度组合（cuboid）进行计算。计算后的 cube 以 key-value 的形式存储在 HBase 中。

对于 SQL 请求，Kylin 的查询引擎会判断 SQL 能否由 HBase 中的 cube 满足。如果可以，查询就会转换为 HBase 的 range scan 操作，直接获得结果数据，因此能做到秒级的响应。

对于无法由 cube 实现的查询，Kylin 可以将查询路由给 Hive 执行，不过当前版本由于性能原因，disable 了路由查询的功能。

Kylin 数据模型

Kylin 将表中的列分为维度列和指标列。在数据导入和查询时相同维度列中的指标会按照对应的聚合函数 (Sum, Count, Min, Max, 精确去重，近似去重，百分位数，TOPN) 进行聚合。

图片描述
在存储到 HBase 时，Cuboid + 维度 会作为 HBase 的 Rowkey , 指标会作为 HBase 的 Value，一般所有指标会在 HBase 的一个列族，每列对应一个指标，但对于较大的去重指标会单独拆分到第 2 个列族。

Kylin 的数据模型的优化主要是根据实际业务场景，配置合适的 Cube 模型来实现的。影响 Cube 性能的配置主要有以下这些。

Aggregation Groups: Cube 中的维度可以划分到多个聚合组中。默认 kylin 会把所有维度放在一个聚合组，如果你很好的了解你的查询模式，那么你可以创建多个聚合组。
Mandatory Dimensions: 必要维度，用于总是出现的维度。
Hierarchy Dimensions: 层级维度，例如 “国家” -> “省” -> “市” 是一个层级。
Joint Dimensions: 联合维度，有些维度往往一起出现，或者它们的基数非常接近（有 1:1 映射关系）。

Rowkeys: 是由维度编码值组成。”Dictionary” （字典）是默认的编码方式；字典只能处理中低基数（少于一千万）的维度。如果维度基数很高（如大于 1 千万), 选择 “false” 然后为维度输入合适的长度，通常是那列的最大长度值；如果超过最大值，会被截断。
Mandatory Cuboids: 维度组合白名单。确保你想要构建的 cuboid 能被构建。

Advanced Dictionaries: “Global Dictionary” 是用于精确计算 COUNT DISTINCT 的字典，它会将一个非 integer 的值转成 integer，以便于 bitmap 进行去重。

“Segment Dictionary” 是另一个用于精确计算 COUNT DISTINCT 的字典，与 Global Dictionary 不同的是，它是基于一个 segment 的值构建的，因此不支持跨 segments 的汇总计算。

Advanced ColumnFamily: 如果有超过一个的 COUNT DISTINCT 或 TopN 度量，你可以将它们放在更多列簇中，以优化与 HBase 的 I/O。

总结

OLAP 和 OLTP 分别解决了数据聚合问题、数据增删改查问题。OLAP 又分为 ROLAP 和 MOLAP 两种。

ROLAP 借助关系型数据库强大的计算能力，现计算数据聚合结果。MOLAP 则是提前预计算数据聚合结果。

MOLAP 的里面开源数据库的代表是 Apache kylin。Apache kylin 通过空间换时间的方式实现了极致的 OLAP 性能，我们可以根据实际的业务场景，适配合适的 Cube 模型来优化 Apache kylin 的查询性能。

21 Doris：MPP架构的ROLAP王者

要成就一件大事业，必须从小事做起。
——列宁

上一小节我们学习 MOLAP 的代表 Kylin。这一小节我们学习一下 ROLAP 代表：Doris。

What is Doris

Doris 是一个 MPP 的 OLAP 系统，主要整合了 Google Mesa（数据模型），Apache Impala（MPP Query Engine) 和 Apache ORCFile (存储格式，编码和压缩) 的技术。

Doris 的系统架构如下，主要分为 FE 和 BE 两个组件。

FE：Frontend，即 Doris 的前端节点。主要负责接收和返回客户端请求、元数据以及集群管理、查询计划生成等工作。
BE：Backend，即 Doris 的后端节点。主要负责数据存储与管理、查询计划执行等工作。
图片描述
如上图，Doris 的整体架构分为两层。多个 FE 组成第一层，提供 FE 的横向扩展和高可用。多个 BE 组成第二层，负责数据存储与管理。本文主要介绍 FE 这一层中，元数据的设计与实现方式。

FE 节点分为 follower 和 observer 两类。各个 FE 之间，通过 bdbje（BerkeleyDB Java Edition）进行 leader 选举，数据同步等工作。
follower 节点通过选举，其中一个 follower 成为 leader 节点，负责元数据的写入操作。当 leader 节点宕机后，其他 follower 节点会重新选举出一个 leader，保证服务的高可用。
observer 节点仅从 leader 节点进行元数据同步，不参与选举。可以横向扩展以提供元数据的读服务的扩展性。
Doris 数据模型

在 Doris 中，数据以表（Table）的形式进行逻辑上的描述。一张表包括行（Row）和列（Column）。Row 即用户的一行数据。Column 用于描述一行数据中不同的字段。

Column 可以分为两大类：Key 和 Value。从业务角度看，Key 和 Value 可以分别对应维度列和指标列。

根据是否对 value 预聚合、key 是否是唯一的将 Doris 数据模型分为：Aggregate、Uniq、Duplicate。

聚合模型

Uniq 可以看做是一种特殊的 Aggregate。Aggregate、Uniq 都属于聚合模型。

假设业务有如下数据表模式：

ColumnName Type AggregationType Comment
user_id LARGEINT 用户 id
date DATE 数据灌入日期
city VARCHAR(20) 用户所在城市
age SMALLINT 用户年龄
sex TINYINT 用户性别
last_visit_date DATETIME REPLACE 用户最后一次访问时间
cost BIGINT SUM 用户总消费
max_dwell_time INT MAX 用户最大停留时间
min_dwell_time INT MIN 用户最小停留时间
如果转换成 Aggregate 模型建表语句则如下（省略建表语句中的 Partition 和 Distribution 信息）

CREATE TABLE IF NOT EXISTS example_db.expamle_tbl
(
user_id LARGEINT NOT NULL COMMENT “用户id”,
date DATE NOT NULL COMMENT “数据灌入日期时间”,
city VARCHAR(20) COMMENT “用户所在城市”,
age SMALLINT COMMENT “用户年龄”,
sex TINYINT COMMENT “用户性别”,
last_visit_date DATETIME REPLACE DEFAULT “1970-01-01 00:00:00” COMMENT “用户最后一次访问时间”,
cost BIGINT SUM DEFAULT “0” COMMENT “用户总消费”,
max_dwell_time INT MAX DEFAULT “0” COMMENT “用户最大停留时间”,
min_dwell_time INT MIN DEFAULT “99999” COMMENT “用户最小停留时间”,
)
AGGREGATE KEY(user_id, date, timestamp, city, age, sex)
… /* 省略 Partition 和 Distribution 信息 */
；
可以看到，这是一个典型的用户信息和访问行为的事实表。在一般星型模型中，用户信息和访问行为一般分别存放在维度表和事实表中。这里我们为了更加方便的解释 Doris 的数据模型，将两部分信息统一存放在一张表中。

表中的列按照是否设置了 AggregationType，分为 Key (维度列) 和 Value（指标列）。没有设置 AggregationType 的，如 user_id、date、age … 等称为 Key，而设置了 AggregationType 的称为 Value。

当我们导入数据时，对于 Key 列相同的行会聚合成一行，而 Value 列会按照设置的 AggregationType 进行聚合。 AggregationType 目前有以下四种聚合方式：

SUM：求和，多行的 Value 进行累加。
REPLACE：替代，下一批数据中的 Value 会替换之前导入过的行中的 Value。
MAX：保留最大值。
MIN：保留最小值。
如果 AggregationType 都为 REPLACE 时候，Aggregate 模型就变成了 Uniq 模型。

Doris 中比较独特的聚合函数是 Replace 函数，这个聚合函数能够保证相同 Keys 的记录只保留最新的 Value，可以借助这个 Replace 函数来实现 点更新。

Uniq 模型模型下，主键是唯一的，如果新导入的数据跟表里面已有数据有相同的主键，新导入的数据会替换表里面已有的数据。

明细模型

在很多分析场景下，我需要分析明细数据，这时候我们需要用到明细模型。

在 Doris 里面 Duplicate 模型是一种明细模型。

举例说明。

ColumnName Type SortKey Comment
timestamp DATETIME Yes 日志时间
type INT Yes 日志类型
error_code INT Yes 错误码
error_msg VARCHAR(1024) No 错误详细信息
op_id BIGINT No 负责人 id
op_time DATETIME No 处理时间
建表语句如下：

CREATE TABLE IF NOT EXISTS example_db.expamle_tbl
(
timestamp DATETIME NOT NULL COMMENT “日志时间”,
type INT NOT NULL COMMENT “日志类型”,
error_code INT COMMENT “错误码”,
error_msg VARCHAR(1024) COMMENT “错误详细信息”,
op_id BIGINT COMMENT “负责人id”,
op_time DATETIME COMMENT “处理时间”
)
DUPLICATE KEY(timestamp, type)
… /* 省略 Partition 和 Distribution 信息 */
；
这种数据模型区别于 Aggregate 和 Uniq 模型。数据完全按照导入文件中的数据进行存储，不会有任何聚合。即使两行数据完全相同，也都会保留。 而在建表语句中指定的 DUPLICATE KEY，只是用来指明底层数据按照那些列进行排序。

Doris Rollup

Doris 中和 Kylin 的 Cuboid 等价的概念是 RollUp 表，**Cuboid 和 RollUp 表都可以认为是一种 Materialized Views 或者 Index **。Doris 的 RollUp 表和 Kylin 的 Cuboid 一样，在查询时不需要显示指定，系统内部会根据查询条件进行路由。

Doris 中 RollUp 表的路由规则如下：

选择包含所有查询列的 RollUp 表
按照过滤和排序的 Column 筛选最符合的 RollUp 表
按照 Join 的 Column 筛选最符合的 RollUp 表
行数最小的
列数最小的
为了大家更好的理解 Doris 的 Rollup，这里将 Rollup 和 Cuboid 进行对比。

Doris 元数据结构

Doris 的元数据是全内存的。每个 FE 内存中，都维护一个完整的元数据镜像。

元数据在内存中整体采用树状的层级结构存储，并且通过添加辅助结构，能够快速访问各个层级的元数据信息。

Doris 元信息所存储的内容：
有哪些“

Doris 的元数据主要存储 4 类数据：

用户数据信息。包括数据库、表的 Schema、分片信息等。
各类作业信息。如导入作业，Clone 作业、SchemaChange 作业等。
用户及权限信息。
集群及节点信息。
Doris 数据流

元数据的数据流具体过程如下：

只有 leader FE 可以对元数据进行写操作。写操作在修改 leader 的内存后，会序列化为一条 log，按照 key-value 的形式写入 bdbje。其中 key 为连续的整型，作为 log id，value 即为序列化后的操作日志。

日志写入 bdbje 后，bdbje 会根据策略（写多数 / 全写），将日志复制到其他 non-leader 的 FE 节点。non-leader FE 节点通过对日志回放，修改自身的元数据内存镜像，完成与 leader 节点的元数据同步。

leader 节点的日志条数达到阈值后（默认 10w 条），会启动 checkpoint 线程。checkpoint 会读取已有的 image 文件，和其之后的日志，重新在内存中回放出一份新的元数据镜像副本。

然后将该副本写入到磁盘，形成一个新的 image。之所以是重新生成一份镜像副本，而不是将已有镜像写成 image，主要是考虑写 image 加读锁期间，会阻塞写操作。所以每次 checkpoint 会占用双倍内存空间。

image 文件生成后，leader 节点会通知其他 non-leader 节点新的 image 已生成。non-leader 主动通过 http 拉取最新的 image 文件，来更换本地的旧文件。

bdbje 中的日志，在 image 做完后，会定期删除旧的日志。

总结

Doris 是从百度内部自主研发并贡献到 Apache 开源社区的 ROLAP 数据库。

Doris 整合了 Google Mesa（数据模型），Apache Impala（MPP Query Engine) 和 Apache ORCFile (存储格式，编码和压缩) 技术，在数据查询延迟上表现非常突出。

Doris 的聚合模型主要用于数据的汇总分析，明细模型主要用于明细数据的查询。相对于 Kylin 只支持汇总模型，Doris 适用的数据场景更加广泛。