shuffle是大数据计算过程中最神奇的地方，不管是MapReduce还是Spark，只要是大数据批处理计算，一定都会有shuffle过程，只有让数据关联起来，数据的内在关系和价值才会呈现出来。如果你不理解shuffle，肯定会在map和reduce编程中产生困惑，不知道该如何正确设计map的输出和reduce的输入。

资源调度框架Yarn

在MapReduce应用程序的启动过程中，最重要的就是要把MapReduce程序分发到大数据集群的服务器上，在Hadoop 1中，这个过程主要是通过TaskTracker和JobTracker通信来完成。

Yarn包括两个部分：一个是资源管理器（Resource Manager），一个是节点管理器（Node Manager）。这也是Yarn的两种主要进程：ResourceManager进程负责整个集群的资源调度管理，通常部署在独立的服务器上；NodeManager进程负责具体服务器上的资源和任务管理，在集群的每一台计算服务器上都会启动，基本上跟HDFS的DataNode进程一起出现。

具体说来，资源管理器又包括两个主要组件：调度器和应用程序管理器。

调度器其实就是一个资源分配算法，根据应用程序（Client）提交的资源申请和当前服务器集群的资源状况进行资源分配。Yarn内置了几种资源调度算法，包括Fair Scheduler、Capacity Scheduler等，你也可以开发自己的资源调度算法供Yarn调用。

Yarn进行资源分配的单位是容器（Container），每个容器包含了一定量的内存、CPU等计算资源，默认配置下，每个容器包含一个CPU核心。容器由NodeManager进程启动和管理，NodeManger进程会监控本节点上容器的运行状况并向ResourceManger进程汇报。

应用程序管理器负责应用程序的提交、监控应用程序运行状态等。应用程序启动后需要在集群中运行一个ApplicationMaster，ApplicationMaster也需要运行在容器里面。每个应用程序启动后都会先启动自己的ApplicationMaster，由ApplicationMaster根据应用程序的资源需求进一步向ResourceManager进程申请容器资源，得到容器以后就会分发自己的应用程序代码到容器上启动，进而开始分布式计算

Yarn的整个工作流程。

1.向Yarn提交应用程序，包括MapReduce ApplicationMaster、编写的MapReduce程序，以及MapReduce Application启动命令。

2.ResourceManager进程和NodeManager进程通信，根据集群资源，为用户程序分配第一个容器，并将MapReduce ApplicationMaster分发到这个容器上面，并在容器里面启动MapReduce ApplicationMaster。

3.MapReduce ApplicationMaster启动后立即向ResourceManager进程注册，并为自己的应用程序申请容器资源。

4.MapReduce ApplicationMaster申请到需要的容器后，立即和相应的NodeManager进程通信，将用户MapReduce程序分发到NodeManager进程所在服务器，并在容器中运行，运行的就是Map或者Reduce任务。

5.Map或者Reduce任务在运行期和MapReduce ApplicationMaster通信，汇报自己的运行状态，如果运行结束，MapReduce ApplicationMaster向ResourceManager进程注销并释放所有的容器资源。

MapReduce如果想在Yarn上运行，就需要开发遵循Yarn规范的MapReduce ApplicationMaster，相应地，其他大数据计算框架也可以开发遵循Yarn规范的ApplicationMaster，这样在一个Yarn集群中就可以同时并发执行各种不同的大数据计算框架，实现资源的统一调度管理。

Hadoop的三个主要组成部分的时候，管HDFS叫分布式文件系统，管MapReduce叫分布式计算框架，管Yarn叫分布式集群资源调度框架。

为什么HDFS是系统，而MapReduce和Yarn则是框架？

框架在架构设计上遵循一个重要的设计原则叫“依赖倒转原则”，依赖倒转原则是高层模块不能依赖低层模块，它们应该共同依赖一个抽象，这个抽象由高层模块定义，由低层模块实现。

应用程序启动器ASM始终运行在 ResourceManager 进程内部。AM运行在集群中任意一个 DataNode（NodeManager） 的 Container 里

用户：提交一个 Java JAR 包给 YARN。

ASM (在 RM 内部)：收到请求，检查资源。

ASM (在 RM 内部)：找 Scheduler 要了一块地（Container）。

ASM (在 RM 内部)：联系某个 NodeManager，说：“在那块地里把这个任务的 ApplicationMaster (AM) 给我跑起来！”

NodeManager：启动 AM。此时，任务正式开始。

AM：启动后，再去直接找 Scheduler 要更多的资源跑具体的 Map/Reduce 任务。

Hive与MapReduce的结合

Hive的目的就是将SQL生成MapReduce可执行的代码，然后提交Hadoop执行.

Spark

RDD是Spark的核心概念，是弹性数据集（Resilient Distributed Datasets）的缩写。RDD既是Spark面向开发者的编程模型，又是Spark自身架构的核心元素。

大数据计算就是在大规模的数据集上进行一系列的数据计算处理。MapReduce针对输入数据，将计算过程分为两个阶段，一个Map阶段，一个Reduce阶段，可以理解成是面向过程的大数据计算。在用MapReduce编程的时候，思考的是，如何将计算逻辑用Map和Reduce两个阶段实现，map和reduce函数的输入和输出是什么

而Spark则直接针对数据进行编程，将大规模数据集合抽象成一个RDD对象，然后在这个RDD上进行各种计算处理，得到一个新的RDD，继续计算处理，直到得到最后的结果数据。所以Spark可以理解成是面向对象的大数据计算。我们在进行Spark编程的时候，思考的是一个RDD对象需要经过什么样的操作，转换成另一个RDD对象，思考的重心和落脚点都在RDD上。

RDD上定义的函数分两种，一种是转换（transformation）函数，这种函数的返回值还是RDD；另一种是执行（action）函数，这种函数不再返回RDD。

RDD定义了很多转换操作函数，比如有计算map(func)、过滤filter(func)、合并数据集union(otherDataset)、根据Key聚合reduceByKey(func, [numPartitions])、连接数据集join(otherDataset, [numPartitions])、分组groupByKey([numPartitions])等十几个函数。

跟MapReduce一样，Spark也是对大数据进行分片计算，Spark分布式计算的数据分片、任务调度都是以RDD为单位展开的，每个RDD分片都会分配到一个执行进程去处理。

RDD上的转换操作又分成两种，一种转换操作产生的RDD不会出现新的分片，比如map、filter等，也就是说一个RDD数据分片，经过map或者filter转换操作后，结果还在当前分片。就像你用map函数对每个数据加1，得到的还是这样一组数据，只是值不同。实际上，Spark并不是按照代码写的操作顺序去生成RDD，比如rdd2 = rdd1.map(func)这样的代码并不会在物理上生成一个新的RDD。物理上，Spark只有在产生新的RDD分片时候，才会真的生成一个RDD，Spark的这种特性也被称作惰性计算。另一种转换操作产生的RDD则会产生新的分片，比如reduceByKey，来自不同分片的相同Key必须聚合在一起进行操作，这样就会产生新的RDD分片

Spark为什么更高效

和MapReduce一样，Spark也遵循移动计算比移动数据更划算这一大数据计算基本原则。但是和MapReduce僵化的Map与Reduce分阶段计算相比，Spark的计算框架更加富有弹性和灵活性，进而有更好的运行性能。

和MapReduce一个应用一次只运行一个map和一个reduce不同，Spark可以根据应用的复杂程度，分割成更多的计算阶段（stage），这些计算阶段组成一个有向无环图DAG，Spark任务调度器可以根据DAG的依赖关系执行计算阶段。

Spark作业调度执行的核心是DAG,DAG根据应用输入划分不同阶段。整个应用就被切分成哪些阶段，每个阶段的依赖关系也就清楚了。之后再根据每个阶段要处理的数据量生成相应的任务集合（TaskSet），每个任务都分配一个任务进程去处理，Spark就实现了大数据的分布式计算。

那么Spark划分计算阶段的依据是什么呢？显然并不是RDD上的每个转换函数都会生成一个计算阶段。

关于计算阶段的划分从图上就能看出规律，当RDD之间的转换连接线呈现多对多交叉连接的时候，就会产生新的阶段。一个RDD代表一个数据集，图中每个RDD里面都包含多个小块，每个小块代表RDD的一个分片。

一个数据集中的多个数据分片需要进行分区传输，写入到另一个数据集的不同分片中，这种数据分区交叉传输的操作，我们在MapReduce的运行过程中也看到过。Spark也需要通过shuffle将数据进行重新组合，相同Key的数据放在一起，进行聚合、关联等操作，因而每次shuffle都产生新的计算阶段。这也是为什么计算阶段会有依赖关系，它需要的数据来源于前面一个或多个计算阶段产生的数据，必须等待前面的阶段执行完毕才能进行shuffle，并得到数据。

不同阶段的依赖关系是有向的，计算过程只能沿着依赖关系方向执行，被依赖的阶段执行完成之前，依赖的阶段不能开始执行，同时，这个依赖关系不能有环形依赖，否则就成为死循环了。

RDDB是窄依赖，而RDDF是宽依赖

因为RDD B在前面一个阶段，阶段1的shuffle过程中，已经进行了数据分区。分区数目和分区Key不变，就不需要再进行shuffle。

计算阶段划分的依据是shuffle，不是转换函数的类型，有的函数有时候有shuffle，有时候没有。

这种不需要进行shuffle的依赖，在Spark里被称作窄依赖；相反的，需要进行shuffle的依赖，被称作宽依赖。跟MapReduce一样，shuffle也是Spark最重要的一个环节，只有通过shuffle，相关数据才能互相计算，构建起复杂的应用逻辑。

同样都要经过shuffle，为什么Spark可以更高效呢？

其实从本质上看，Spark可以算作是一种MapReduce计算模型的不同实现。Hadoop MapReduce简单粗暴地根据shuffle将大数据计算分成Map和Reduce两个阶段，然后就算完事了。而Spark更细腻一点，将前一个的Reduce和后一个的Map连接起来，当作一个阶段持续计算，形成一个更加优雅、高效的计算模型，虽然其本质依然是Map和Reduce。但是这种多个计算阶段依赖执行的方案可以有效减少对HDFS的访问，减少作业的调度执行次数，因此执行速度也更快。

并且和Hadoop MapReduce主要使用磁盘存储shuffle过程中的数据不同，Spark优先使用内存进行数据存储，包括RDD数据。除非是内存不够用了，否则是尽可能使用内存， 这也是Spark性能比Hadoop高的另一个原因

总结，Spark为什么更快: 1. MapReduce模型不完全一样，Spark将前一个Reduce而后一个Map连接起来，将一个计算任务拆为多个阶段,减少了对HDFS的访问和作业的调度执行次数。2. Spark优先使用内存，除非内存不够用。而Hadoop在每次Map完后都会将数据写入到磁盘.

Spark里面的RDD函数有两种，一种是转换函数，调用以后得到的还是一个RDD，RDD的计算逻辑主要通过转换函数完成。

另一种是action函数，调用以后不再返回RDD。比如count()函数，返回RDD中数据的元素个数；saveAsTextFile(path)，将RDD数据存储到path路径下。Spark的DAGScheduler在遇到shuffle的时候，会生成一个计算阶段，在遇到action函数的时候，会生成一个作业（job）。

RDD里面的每个数据分片，Spark都会创建一个计算任务去处理，所以一个计算阶段会包含很多个计算任务（task）。

Spark的执行过程

Spark支持Standalone、Yarn、Mesos、Kubernetes等多种部署方案，几种部署方案原理也都一样，只是不同组件角色命名不同，但是核心功能和运行流程都差不多。

首先，Spark应用程序启动在自己的JVM进程里，即Driver进程，启动后调用SparkContext初始化执行配置和输入数据。SparkContext启动DAGScheduler构造执行的DAG图，切分成最小的执行单位也就是计算任务。

首先，Spark应用程序启动在自己的JVM进程里，即Driver进程，启动后调用SparkContext初始化执行配置和输入数据。SparkContext启动DAGScheduler构造执行的DAG图，切分成最小的执行单位也就是计算任务。

然后Driver向Cluster Manager请求计算资源，用于DAG的分布式计算。Cluster Manager收到请求以后，将Driver的主机地址等信息通知给集群的所有计算节点Worker。

Worker收到信息以后，根据Driver的主机地址，跟Driver通信并注册，然后根据自己的空闲资源向Driver通报自己可以领用的任务数。Driver根据DAG图开始向注册的Worker分配任务。

Worker收到任务后，启动Executor进程开始执行任务。Executor先检查自己是否有Driver的执行代码，如果没有，从Driver下载执行代码，通过Java反射加载后开始执行。

Spark有三个主要特性：RDD的编程模型更简单，DAG切分的多阶段计算过程更快速，使用内存存储中间计算结果更高效。这三个特性使得Spark相对Hadoop MapReduce可以有更快的执行速度，以及更简单的编程实现。

BigTable对应的NOSQL

Google发表GFS、MapReduce、BigTable三篇论文，号称“三驾马车”，开启了大数据的时代。那和这“三驾马车”对应的有哪些开源产品呢？GFS对应的Hadoop分布式文件系统HDFS，以及MapReduce对应的Hadoop分布式计算框架MapReduce。而BigTable就对应NOSQL数据库 HBase.

HBase为可伸缩海量数据储存而设计，实现面向在线业务的实时数据访问延迟。HBase的伸缩性主要依赖其可分裂的HRegion及可伸缩的分布式文件系统HDFS实现

在计算机数据存储领域，一直是关系数据库（RDBMS）的天下，以至于在传统企业的应用领域，许多应用系统设计都是面向数据库设计，也就是先设计数据库然后设计程序，从而导致关系模型绑架对象模型,并由此引申出旷日持久的业务对象贫血模型与充血模型之争。

业务的贫血模型和充血模型

贫血模型 这是目前国内互联网开发中最常用的模式（常见于 Spring 的三层架构：Controller-Service-Dao）。业务对象（POJO/Entity）只包含属性（Getter/Setter），没有任何业务逻.所有的业务逻辑都写在 Service 层中。

充血模型 (Rich Domain Model)这是领域驱动设计（DDD）极力推崇的模式。业务对象不仅包含属性，还包含与该对象相关的业务行为（方法）。业务逻辑封装在领域对象内部。Service 只负责编排这些对象。

关系数据库难以克服的缺陷——糟糕的海量数据处理能力及僵硬的设计约束，从Google的BigTable开始，一系列的可以进行海量数据存储与访问的数据库被设计出来，更进一步说，NoSQL这一概念被提了出来。

HBase的架构设计

HBase之所以能够具有海量数据处理能力，其根本在于和传统关系型数据库设计的不同思路。传统关系型数据库对存储在其上的数据有很多约束，学习关系数据库都要学习数据库设计范式，事实上，是在数据存储中包含了一部分业务逻辑。而NoSQL数据库则简单暴力地认为，数据库就是存储数据的，业务逻辑应该由应用程序去处理

HRegion是HBase负责数据存储的主要进程，应用程序对数据的读写操作都是通过和HRegion通信完成。在HBase中，数据以HRegion为单位进行管理，也就是说应用程序如果想要访问一个数据，必须先找到HRegion，然后将数据读写操作提交给HRegion，由 HRegion完成存储层面的数据操作。

可伸缩架构

HRegionServer是物理服务器，每个HRegionServer上可以启动多个HRegion实例。当一个 HRegion中写入的数据太多，达到配置的阈值时，一个HRegion会分裂成两个HRegion，并将HRegion在整个集群中进行迁移，以使HRegionServer的负载均衡。

每个HRegion中存储一段Key值区间[key1, key2)的数据，所有HRegion的信息，包括存储的Key值区间、所在HRegionServer地址、访问端口号等，都记录在HMaster服务器上。为了保证HMaster的高可用，HBase会启动多个HMaster，并通过ZooKeeper选举出一个主服务器。

应用程序通过ZooKeeper获得主HMaster的地址，输入Key值获得这个Key所在的HRegionServer地址，然后请求HRegionServer上的HRegion，获得所需要的数据。

数据写入过程也是一样，需要先得到HRegion才能继续操作。HRegion会把数据存储在若干个HFile格式的文件中，这些文件使用HDFS分布式文件系统存储，在整个集群内分布并高可用。当一个HRegion中数据量太多时，这个HRegion连同HFile会分裂成两个HRegion，并根据集群中服务器负载进行迁移。

如果集群中有新加入的服务器，也就是说有了新的HRegionServer，由于其负载较低，也会把HRegion迁移过去并记录到HMaster，从而实现HBase的线性伸缩。HBase的核心设计目标是解决海量数据的分布式存储，和Memcached这类分布式缓存的路由算法不同，HBase的做法是按Key的区域进行分片，这个分片也就是HRegion。应用程序通过HMaster查找分片，得到HRegion所在的服务器HRegionServer，然后和该服务器通信，就得到了需要访问的数据。

可扩展数据模型

传统的关系数据库为了保证关系运算的正确性，在设计数据库表结构的时候，需要指定表的schema也就是字段名称、数据类型等，并要遵循特定的设计范式。这些规范带来了一个问题，就是僵硬的数据结构难以面对需求变更带来的挑战，有些应用系统设计者通过预先设计一些冗余字段来应对，但显然这种设计也很糟糕。

许多NoSQL数据库使用的列族（ColumnFamily）做到可扩展的护具结构设计。列族最早在Google的BigTable中使用，这是一种面向列族的稀疏矩阵存储格式，

高性能存储

传统的机械式磁盘的访问特性是连续读写很快，随机读写很慢。这是因为机械磁盘靠电机驱动访问磁盘上的数据，电机要将磁头落到数据所在的磁道上，这个过程需要较长的寻址时间。如果数据不连续存储，磁头就要不停地移动，浪费了大量的时间。

虽然 SSD 很快，但它在物理层面有一个非常“别扭”的限制，这导致它依然存在顺序写比随机写快的现象：

• 读取（Read）：以 Page（页，约 4KB~16KB） 为单位，非常快。
• 写入（Write）：也以 Page 为单位。但有个前提：必须在“干净”的空地上写。
• 擦除（Erase）：以 Block（块，包含数百个 Page） 为单位。

核心矛盾：你不能直接覆盖旧数据。如果你想改写一个 Page，必须先把整个 Block 擦除干净，然后才能写。这种特性被称为 “写前擦除”（Erase-before-write）。

为了提高数据写入速度，HBase使用了一种叫作LSM树的数据结构进行数据存储。LSM树的全名是Log Structed Merge Tree，翻译过来就是Log结构合并树。数据写入的时候以Log方式连续写入，然后异步对磁盘上的多个LSM树进行合并。

LSM 树并不是一棵像 B+ 树那样巨大的、常驻磁盘的树，它实际上是由内存组件和磁盘组件共同构成的多层结构。

LSM树可以看作是一个N阶合并树。数据写操作（包括插入、修改、删除）都在内存中进行，并且都会创建一个新记录（修改会记录新的数据值，而删除会记录一个删除标志）。这些数据在内存中仍然还是一棵排序树，当数据量超过设定的内存阈值后，会将这棵排序树和磁盘上最新的排序树合并。当这棵排序树的数据量也超过设定阈值后，会和磁盘上下一级的排序树合并。合并过程中，会用最新更新的数据覆盖旧的数据（或者记录为不同版本）。

在需要进行读操作时，总是从内存中的排序树开始搜索，如果没有找到，就从磁盘 上的排序树顺序查找。

在LSM树上进行一次数据更新不需要磁盘访问，在内存即可完成。当数据访问以写操作为主，而读操作则集中在最近写入的数据上时，使用LSM树可以极大程度地减少磁盘的访问次数，加快访问速度。

HBase作为Google BigTable的开源实现，完整地继承了BigTable的优良设计。架构上通过数据分片的设计配合HDFS，实现了数据的分布式海量存储；数据结构上通过列族的设计，实现了数据表结构可以在运行期自定义；存储上通过LSM树的方式，使数据可以通过连续写磁盘的方式保存数据，极大地提高了数据写入性能。

流式计算

大数据批处理计算。顾名思义，数据是以批为单位进行计算，比如一天的访问日志、历史上所有的订单数据等。这些数据通常通过HDFS存储在磁盘上，使用MapReduce或者Spark这样的批处理大数据计算框架进行计算，一般完成一次计算需要花费几分钟到几小时的时间。

此外，还有一种大数据技术，针对实时产生的大规模数据进行即时计算处理，我们比较熟悉的有摄像头采集的实时视频数据、淘宝实时产生的订单数据等。像上海这样的一线城市，公共场所的摄像头规模在数百万级，即使只有重要场所的视频数据需要即时处理，可能也会涉及几十万个摄像头，如果想实时发现视频中出现的通缉犯或者违章车辆，就需要对这些摄像头产生的数据进行实时处理。实时处理最大的不同就是这类数据跟存储在HDFS上的数据不同，是实时传输过来的，或者形象地说是流过来的，所以针对这类大数据的实时处理系统也叫大数据流计算系统。

Storm

能不能开发一个流处理计算系统，我们只要定义好处理流程和每一个节点的处理逻辑，代码部署到流处理系统后，就能按照预定义的处理流程和处理逻辑执行呢？Storm就是在这种背景下产生的，它也算是一个比较早期的大数据流计算框架。上面的例子如果用Storm来实现，过程就变得简单一些了。

Storm运行机制

nimbus是集群的Master，负责集群管理、任务分配等。supervisor是Slave，是真正完成计算的地方，每个supervisor启动多个worker进程，每个worker上运行多个task，而task就是spout或者bolt。supervisor和nimbus通过ZooKeeper完成任务分配、心跳检测等操作。

nimbus是集群的Master，负责集群管理、任务分配等。supervisor是Slave，是真正完成计算的地方，每个supervisor启动多个worker进程，每个worker上运行多个task，而task就是spout或者bolt。supervisor和nimbus通过ZooKeeper完成任务分配、心跳检测等操作。

Hadoop、Storm的设计理念，其实是一样的，就是把和具体业务逻辑无关的东西抽离出来，形成一个框架，比如大数据的分片处理、数据的流转、任务的部署与执行等，开发者只需要按照框架的约束，开发业务逻辑代码，提交给框架执行就可以了。

而这也正是所有框架的开发理念，就是将业务逻辑和处理过程分离开来，使开发者只需关注业务开发即可，比如Java开发者都很熟悉的Tomcat、Spring等框架，全部都是基于这种理念开发出来的

Spark Streaming

Spark Streaming巧妙地利用了Spark的分片和快速计算的特性，将实时传输进来的数据按照时间进行分段，把一段时间传输进来的数据合并在一起，当作一批数据，再去交给Spark去处理。

如果时间段分得足够小，每一段的数据量就会比较小，再加上Spark引擎的处理速度又足够快，这样看起来好像数据是被实时处理的一样，Spark Streaming主要负责将流数据转换成小的批数据，剩下的就可以交给Spark去做了。

Flink

Flink的架构和Hadoop 1或者Yarn看起来也很像，JobManager是Flink集群的管理者，Flink程序提交给JobManager后，JobManager检查集群中所有TaskManager的资源利用状况，如果有空闲TaskSlot（任务槽），就将计算任务分配给它执行。

如果要进行流计算，Flink会初始化一个流执行环境StreamExecutionEnvironment，然后利用这个执行环境构建数据流DataStream。

1
2
3

StreamExecutionEnvironment see = StreamExecutionEnvironment.getExecutionEnvironment();

DataStream<WikipediaEditEvent> edits = see.addSource(new WikipediaEditsSource());

如果要进行批处理计算，Flink会初始化一个批处理执行环境ExecutionEnvironment，然后利用这个环境构建数据集DataSet。

1
2
3

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

DataSet<String> text = env.readTextFile("/path/to/file");

然后在DataStream或者DataSet上执行各种数据转换操作（transformation），这点很像Spark。不管是流处理还是批处理，Flink运行时的执行引擎是相同的，只是数据源不同而已。

Zookeeper如何保证数据一致性

在分布式系统里的多台服务器要对数据状态达成一致，其实是一件很有难度和挑战的事情，因为服务器集群环境的软硬件故障随时会发生，多台服务器对一个数据的记录保持一致，需要一些技巧和设计。

HDFS为了保证整个集群的高可用，需要部署两台NameNode服务器，一台作为主服务器，一台作为从服务器。当主服务器不可用的时候，就切换到从服务器上访问。但是如果不同的应用程序（Client）或者DataNode做出的关于主服务器是否可用的判断不同，那么就会导致HDFS集群混乱。

CAP原理认为，一个提供数据服务的分布式系统无法同时满足数据一致性（Consistency）、可用性（Availibility）、分区耐受性（Patition Tolerance）这三个条件，

一致性是说，每次读取的数据都应该是最近写入的数据或者返回一个错误（Every read receives the most recent write or an error），而不是过期数据，也就是说，数据是一致的。

可用性是说，每次请求都应该得到一个响应，而不是返回一个错误或者失去响应，不过这个响应不需要保证数据是最近写入的（Every request receives a (non-error) response, without the guarantee that it contains the most recent write），也就是说系统需要一直都是可以正常使用的，不会引起调用者的异常，但是并不保证响应的数据是最新的。

分区耐受性是说，即使因为网络原因，部分服务器节点之间消息丢失或者延迟了，系统依然应该是可以操作的（The system continues to operate despite an arbitrary number of messages being dropped (or delayed) by the network between nodes）。

当网络分区失效发生的时候，我们要么取消操作，这样数据就是一致的，但是系统却不可用；要么我们继续写入数据，但是数据的一致性就得不到保证。

对于一个分布式系统而言，网络失效一定会发生，也就是说，分区耐受性是必须要保证的，那么在可用性和一致性上就必须二选一。当网络分区失效，也就是网络不可用的时候，如果选择了一致性，系统就可能返回一个错误码或者干脆超时，即系统不可用。如果选择了可用性，那么系统总是可以返回一个数据，但是并不能保证这个数据是最新的。

Zookeeper架构与ZAB算法

ZooKeeper主要提供数据的一致性服务，其实现分布式系统的状态一致性依赖一个叫Paxos的算法。Paxos算法在多台服务器通过内部的投票表决机制决定一个数据的更新与写入。

应用程序连接到任意一台服务器后提起状态修改请求（也可以是获得某个状态锁的请求），从图上看也就是服务器1，会将这个请求发送给集群中其他服务器进行表决。如果某个服务器同时收到了另一个应用程序同样的修改请求，它可能会拒绝服务器1的表决，并且自己也发起一个同样的表决请求，那么其他服务器就会根据时间戳和服务器排序规则进行表决。

表决结果会发送给其他所有服务器，最终发起表决的服务器也就是服务器1，会根据收到的表决结果决定该修改请求是否可以执行，事实上，只有在收到多数表决同意的情况下才会决定执行。当有多个请求同时修改某个数据的情况下，服务器的表决机制保证只有一个请求会通过执行，从而保证了数据的一致性。

Paxos算法有点过于复杂、实现难度也比较高，所以ZooKeeper在编程实现的时候将其简化成了一种叫做ZAB的算法（Zookeeper Atomic Broadcast， Zookeeper原子广播）。

ZAB算法的目的，同样是在多台服务器之间达成一致，保证这些服务器上存储的数据是一致的。ZAB算法的主要特点在于：需要在这些服务器中选举一个Leader，所有的写请求都必须提交给Leader。由Leader服务器向其他服务器（Follower）发起Propose，通知所有服务器：我们要完成一个写操作请求，大家检查自己的数据状态，是否有问题。

如果所有Follower服务器都回复Leader服务器ACK，即没有问题，那么Leader服务器会向所有Follower发送Commit命令，要求所有服务器完成写操作。这样包括Leader服务器在内的所有ZooKeeper集群服务器的数据，就都更新并保持一致了。如果有两个客户端程序同时请求修改同一个数据，因为必须要经过Leader的审核，而Leader只接受其中一个请求，数据也会保持一致。

在实际应用中，客户端程序可以连接任意一个Follower，进行数据读写操作。如果是写操作，那么这个请求会被这个Follower发送给Leader，进行如上所述的处理；如果是读操作，因为所有服务器的数据都是一致的，那么这个Follower直接返回自己本地的数据给客户端就可以了。

对数据进行分类和预测

上面通过Hadoop,Spark,Flink等对大数据进行存储，计算，查询等分布式计算框架介绍，有了数据之后如何利用呢。 利用数学统计方法，统计数据中的规律，然后利用这些统计规律进行自动化数据处理，使计算机表现出某种智能的特性，而各种数学统计方法，就是大数据算法。

KNN分类算法

KNN算法，即K近邻（K Nearest Neighbour）算法，是一种基本的分类算法。其主要原理是：对于一个需要分类的数据，将其和一组已经分类标注好的样本集合进行比较，得到距离最近的K个样本，K个样本最多归属的类别，就是这个需要分类数据的类别

数据的距离

KNN算法的关键是要比较需要分类的数据与样本数据之间的距离，这在机器学习中通常的做法是：提取数据的特征值，根据特征值组成一个n维实数向量空间（这个空间也被称作特征空间），然后计算向量之间的空间距离。空间之间的距离计算方法有很多种，常用的有欧氏距离、余弦距离等。

文本的特征值

机器学习的算法需要计算距离，而计算距离还需要知道数据的特征向量，因此提取数据的特征向量是机器学习工程师们的重要工作，有时候甚至是最重要的工作。不同的数据以及不同的应用场景需要提取不同的特征值，我们以比较常见的文本数据为例，看看如何提取文本特征向量。

文本数据的特征值就是提取文本关键词，TF-IDF算法是比较常用且直观的一种文本关键词提取算法。这种算法是由TF和IDF两部分构成。

TF是词频（Term Frequency），表示某个单词在文档中出现的频率，一个单词在一个文档中出现得越频繁，TF值越高。

词频：

IDF是逆文档频率（Inverse Document Frequency），表示这个单词在所有文档中的稀缺程度，越少文档出现这个词，IDF值越高。

逆文档频率：

TF与IDF的乘积就是TF-IDF。

贝叶斯公式

为了理解这个公式，我们给每一个项起个名字：

• $P(A|B)$（后验概率）：在看到证据 $B$ 之后，事件 $A$ 发生的概率。
• $P(A)$（先验概率）：在看到证据 $B$ 之前，我们对事件 $A$ 发生概率的初始猜测。
• $P(B|A)$（似然度）：如果假设 $A$ 是真的，那么证据 $B$ 出现的可能性有多大。
• $P(B)$（标准化常量）：证据 $B$ 在所有情况下出现的总概率。

可以把贝叶斯公式看成一个“认知自动机”：

初始信念 ($P(A)$) + 新的证据 ($P(B|A)$) $\rightarrow$ 更新后的信念 ($P(A|B)$)

数据之间的关系

搜索排序

当我们使用Google进行搜索的时候，通常在搜索的前三个结果里就能找到自己想要的网页内容，而且很大概率第一个结果就是我们想要的网页。而排名越往后，搜索结果与我期望的偏差越大。并且在搜索结果页的上面，会提示总共找到多少个结果。

那么Google为什么能在十几万的网页中知道我最想看的网页是哪些，然后把这些页面排到最前面呢？

答案是Google使用了一种叫PageRank的算法，这种算法根据网页的链接关系给网页打分。如果一个网页A，包含另一个网页B的超链接，那么就认为A网页给B网页投了一票，PageRank算法核心包括数量和质量

这样经过一次计算后，每个页面的PageRank分值就会重新分配，重复同样的算法过程，经过几次计算后，根据每个页面PageRank分值进行排序，就得到一个页面重要程度的排名表。根据这个排名表，将用户搜索出来的网页结果排序，排在前面的通常也正是用户想要的结果。

但是这个算法还有个问题，如果某个页面只包含指向自己的超链接，这样的话其他页面不断给它送分，而自己一分不出，随着计算执行次数越多，它的分值也就越高，这显然是不合理的。这种情况就像下图所示的，A页面只包含指向自己的超链接。

Google的解决方案是，设想浏览一个页面的时候，有一定概率不是点击超链接，而是在地址栏输入一个URL访问其他页面，表示在公式上，就是

上面$（1-\alpha）$就是跳转到其他任何页面的概率，通常取经验值0.15（即$\alpha$ 为0.85），因为有一定概率输入的URL是自己的，所以加上上面公式最后一项，其中分母4表示所有网页的总数。

那么对于$N$个网页，任何一个页面$P_{i}$的PageRank计算公式如下

公式中，$P{j}\in M(P{i})$表示所有包含有$P{i}$超链接的$P{j}$，$L(P{j})$表示$P{j}$页面包含的超链接数，$N$表示所有的网页总和。

由于Google要对全世界的网页进行排名，所以这里的N可能是一个万亿级的数字，一开始将所有页面的PageRank值设为1，带入上面公式计算，每个页面都得到一个新的PageRank值。再把这些新的PageRank值带入上面的公式，继续得到更新的PageRank值，如此迭代计算，直到所有页面的PageRank值几乎不再有大的变化才停止。

关联分析

在深入关联分析前，需要先了解两个基本概念，一个是支持度，一个是置信度。

支持度是指一组频繁模式的出现概率，比如（啤酒，尿不湿）是一组频繁模式，它的支持度是4%，也就是说，在所有订单中，同时出现啤酒和尿不湿这两件商品的概率是4%。

置信度用于衡量频繁模式内部的关联关系，如果出现尿不湿的订单全部都包含啤酒，那么就可以说购买尿不湿后购买啤酒的置信度是100%；如果出现啤酒的订单中有20%包含尿不湿，那么就可以说购买啤酒后购买尿不湿的置信度是20%。

那应该从哪里考虑着手，可以使用最少的计算资源寻找到最小支持度的频繁模式？寻找满足最小支持度的频繁模式经典算法是Apriori算法，Apriori算法的步骤是：

第1步：设置最小支持度阈值。

第2步：寻找满足最小支持度的单件商品，也就是单件商品出现在所有订单中的概率不低于最小支持度。

第3步：从第2步找到的所有满足最小支持度的单件商品中，进行两两组合，寻找满足最小支持度的两件商品组合，也就是两件商品出现在同一个订单中概率不低于最小支持度。

第4步：从第3步找到的所有满足最小支持度的两件商品，以及第2步找到的满足最小支持度的单件商品进行组合，寻找满足最小支持度的三件商品组合。

第5步：以此类推，找到所有满足最小支持度的商品组合。

Apriori算法极大地降低了需要计算的商品组合数目，这个算法的原理是，如果一个商品组合不满足最小支持度，那么所有包含这个商品组合的其他商品组合也不满足最小支持度。所以从最小商品组合，也就是一件商品开始计算最小支持度，逐渐迭代，进而筛选出所有满足最小支持度的频繁模式。

通过关联分析，可以发现看似不相关商品的关联关系，并利用这些关系进行商品营销，一方面可以为用户提供购买便利；另一方面也能提高企业营收。专栏下一期还会讲到更多发现用户兴趣进行推荐的算法。

聚类

分类算法主要解决如何将一个数据分到几个确定类别中的一类里去。分类算法通常需要样本数据训练模型，再利用模型进行数据分类，那么一堆样本数据又如何知道各自的类别呢？样本数据归类一方面可以通过人工手动打标签，另一方面也可以利用算法进行自动归类，即所谓的“聚类”。

K-means是一种在给定分组个数后，能够对数据进行自动归类，即聚类的算法。

第1步：随机在图中取K个种子点，图中K=2，即图中的实心小圆点。

第2步：求图中所有点到这K个种子点的距离，假如一个点离种子点X最近，那么这个点属于X点群。

第3步：对已经分好组的两组数据，分别求其中心点。对于图中二维平面上的数据，求中心点最简单暴力的算法就是对当前同一个分组中所有点的X坐标和Y坐标分别求平均值，得到的就是中心点。

第4步：重复第2步和第3步，直到每个分组的中心点不再移动。这时候，距每个中心点最近的点数据聚类为同一组数据。

K-means算法原理简单，在知道分组个数的情况下，效果非常好，是聚类经典算法。通过聚类分析我们可以发现事物的内在规律：具有相似购买习惯的用户群体被聚类为一组，一方面可以直接针对不同分组用户进行差别营销，线下渠道的话还可以根据分组情况进行市场划分；另一方面可以进一步分析，比如同组用户的其他统计特征还有哪些，并发现一些有价值的模式。

预测用户的喜好

在用户对自己需求相对明确的时候，可以用搜索引擎通过关键字搜索很方便地找到自己需要的信息。但有些时候，搜索引擎并不能完全满足用户对信息发现的需求。一方面，用户有时候其实对自己的需求并不明确，期望系统能主动推荐一些自己感兴趣的内容或商品；另一方面，企业也希望能够通过更多渠道向用户推荐信息和商品，在改善用户体验的同时，提高成交转化率，获得更多营收。而这中间发现用户兴趣和喜好的就是推荐引擎。

主要就是依靠各种推荐算法，常用的推荐算法有：基于人口统计的推荐、基于商品属性的推荐、基于用户的协同过滤推荐、基于商品的协同过滤推荐

基于人口统计的推荐

基于用户的属性进行分类，然后根据同类用户的行为进行推荐

基于人口统计的推荐是相对比较简单的一种推荐算法，根据用户的基本信息进行分类，然后将商品推荐给同类用户。

这种推荐算法也不依赖商品的数据，和要推荐的领域无关，不管是服装还是美食，不管是电影还是旅游目的地，都可以进行推荐，甚至可以混杂在一起进行推荐。

当然也正因为这种推荐算法比较简单，对于稍微精细一点的场景，推荐效果就比较差了。因此，在人口统计信息的基础上，根据用户浏览、购买信息和其他相关信息，进一步细化用户的分类信息，给用户贴上更多的标签，比如家庭成员、婚姻状况、居住地、学历、专业、工作等，即所谓的用户画像，根据用户画像进行更精细的推荐，并进一步把用户喜好当做标签完善用户画像，再利用更完善的用户画像进行推荐，如此不断迭代优化用户画像和推荐质量。

基于商品属性的推荐

基于商品属性的推荐则是将商品的属性进行分类，然后根据用户的历史行为进行推荐。

例如电影A和电影D有相似的属性，被划分为同类商品，如果用户A喜欢电影A，那么就可以向用户A推荐电影D，比如给喜欢《星球大战》的用户推荐《星际迷航》。一般来说，相对于基于人口统计的推荐，基于商品属性的推荐会更符合用户的口味，推荐效果相对更好一点。

但是基于商品属性的推荐需要对商品属性进行全面的分析和建模，难度相对也更大一点，在实践中，一种简单的做法是提取商品描述的关键词和商品的标签作为商品的属性。此外，基于商品属性的推荐依赖用户的历史行为数据，如果是新用户进来，没有历史数据，就没有办法进行推荐了，即存在“冷启动”问题。

基于用户的协同过滤推荐

基于用户的协同过滤推荐是根据用户的喜好进行用户分类，常用的就是KNN算法，寻找和当前用户喜好最相近的K个用户，然后根据这些用户的喜好为当前用户进行推荐

基于用户的协同过滤推荐和基于人口统计的推荐都是将用户分类后，根据同类用户的喜好为当前用户进行推荐。不同的是，基于人口统计的推荐仅仅根据用户的个人信息进行分类，分类的粒度比较大，准确性也较差；而基于用户的协同过滤推荐则根据用户历史喜好进行分类，能够更准确地反映用户的喜好类别，推荐效果也更好一点。今天文章开头举的推荐电影的例子，就是基于用户的协同过滤进行推荐。

基于商品的协同过滤推荐

基于商品的协同过滤推荐是根据用户的喜好对商品进行分类，如果两个商品，喜欢它们的用户具有较高的重叠性，就认为它们的距离相近，划分为同类商品，然后进行推荐

例如，用户A喜欢商品A、商品B和商品D，用户B喜欢商品B、商品C和商品D，那么商品B和商品D的距离最近，划分为同类商品；而用户C喜欢商品B，那么就可以为其推荐商品D。

商品的分类相对用户的分类更为稳定，通常情况下，商品的数目也少于用户的数目，因此使用基于商品的协同过滤推荐，计算量和复杂度小于基于用户的协同过滤推荐。

除了上面这些推荐算法，还有基于模型的推荐，根据用户和商品数据，训练数学模型，然后进行推荐。在实践中，通常会混合应用多种算法进行推荐，特别是大型电商网站，推荐效果每进步一点，都可能会带来巨大的营收转化，如果你经常在网上购物，肯定也能感受电商网站这些年在推荐方面的巨大进步。

互联网运营数据指标

不同的互联网行业关注不同的运营数据，细化来看，复杂的互联网产品关注的运营指标成百上千。但是有一些指标是我们最常用的，这些指标基本反映了运营的核心状态

新增用户数

新增用户数是网站增长性的关键指标，指新增加的访问网站的用户数（或者新下载App的用户数），对于一个处于爆发期的网站，新增用户数会在短期内出现倍增的走势，是网站的战略机遇期，很多大型网站都经历过一个甚至多个短期内用户暴增的阶段。新增用户数有日新增用户数、周新增用户数、月新增用户数等几种统计口径。

用户留存率

新增的用户并不一定总是对网站（App）满意，在使用网站（App）后感到不满意，可能会注销账户（卸载App），这些辛苦获取来的用户就流失掉了。网站把经过一段时间依然没有流失的用户称作留存用户，留存用户数比当期新增用户数就是用户留存率。

1

用户留存率 = 留存用户数 / 当期新增用户数

计算留存有时间窗口，即和当期数据比，3天前新增用户留存的，称作3日留存；相应的，还有5日留存、7日留存等。新增用户可以通过广告、促销、病毒营销等手段获取

一般说来，3日留存率能做到40%以上就算不错了。和用户留存率对应的是用户流失率。

活跃用户数

用户下载注册，但是很少打开产品，表示产品缺乏黏性和吸引力。活跃用户数表示打开使用产品的用户数，根据统计口径不同，有日活跃用户数、月活跃用户数等。提升活跃是网站运营的重要目标，各类App常用推送优惠促销消息给用户的手段促使用户打开产品。

PV

打开产品就算活跃，打开以后是否频繁操作，就用PV这个指标衡量，用户每次点击，每个页面跳转，被称为一个PV（Page View）。PV是网页访问统计的重要指标，在移动App上，需要进行一些变通来进行统计。

GMV

GMV即成交总金额（Gross Merchandise Volume），是电商网站统计营业额（流水）、反映网站营收能力的重要指标。和GMV配合使用的还有订单量（用户下单总量）、客单价（单个订单的平均价格）等。

转化率

转化率是指在电商网站产生购买行为的用户与访问用户之比。

1

转化率 = 有购买行为的用户数 / 总访问用户数

用户从进入网站（App）到最后购买成功，可能需要经过复杂的访问路径，每个环节都有可能会离开：进入首页想了想没什么要买的，然后离开；搜索结果看了看不想买，然后离开；进入商品详情页面，看看评价、看看图片、看看价格，然后离开；放入购物车后又想了想自己的钱包，然后离开；支付的时候发现不支持自己喜欢的支付方式，然后离开…一个用户从进入网站到支付，完成一笔真正的消费，中间会有很大概率流失，网站必须要想尽各种办法：个性化推荐、打折促销、免运费、送红包、分期支付，以留住用户，提高转化率。

A/B测试与灰度发布

A/B测试将每一次测试当作一个实验。通过A/B测试系统的配置，将用户随机分成两组（或者多组），每组用户访问不同版本的页面或者执行不同的处理逻辑，即运行实验。通常将原来产品特性当作一组，即原始组；新开发的产品特性当作另一组，即测试组。

经过一段时间（几天甚至几周）以后，对A/B测试实验进行分析，观察两组用户的数据指标，使用新特性的测试组是否好于作为对比的原始组，如果效果比较好，那么这个新开发的特性就会在下次产品发布的时候正式发布出去，供所有用户使用；如果效果不好，这个特性就会被放弃，实验结束。

A/B测试的系统架构

A/B测试系统最重要的是能够根据用户ID（或者设备ID）将实验配置参数分发给应用程序，应用程序根据配置参数决定给用户展示的界面和执行的业务逻辑

在实验管理模块里进行用户分组，比如测试组、原始组，并指定每个分组用户占总用户的百分比；流量分配模块根据某种Hash算法将用户（设备）分配到某个实验组中；一个实验可以有多个参数，每个组有不同的参数值。

移动App在启动后，定时和A/B测试系统通信，根据自身用户ID或者设备ID获取自己参与的A/B测试实验的配置项，根据配置项执行不同的代码，体验不同的应用特性。应用服务器和A/B测试系统在同一个数据中心，获取实验配置的方式可以更灵活。

移动App和应用服务器上报实验数据其实就是传统的数据采集，但是在有A/B测试的情况下，数据采集上报的时候需要将A/B测试实验ID和分组ID也上报，然后在数据分析的时候，才能够将同一个实验的不同分组数据分别统计，得到A/B测试的实验数据报告。

灰度发布

经过A/B测试验证过的功能特性，就可以发布到正式的产品版本中，向所有用户开放。但是有时候在A/B测试中表现不错的特性，正式版本发布后效果却不好。此外，A/B测试的时候，每个功能都应该是独立（正交）的，正式发布的时候，所有的特性都会在同一个版本中一起发布，这些特性之间可能会有某种冲突，导致发布后的数据不理想。

解决这些问题的手段是灰度发布，即不是一次将新版本发布给全部用户，而是一批一批逐渐发布给用户。在这个过程中，监控产品的各项数据指标，看是否符合预期，如果数据表现不理想，就停止灰度发布，甚至进行灰度回滚，让所有用户都恢复到以前的版本，进一步观察分析数据指标。

灰度发布系统可以用A/B测试系统来承担，创建一个名叫灰度发布的实验即可，这个实验包含这次要发布的所有特性的参数，然后逐步增加测试组的用户数量，直到占比达到总用户量的100%，即为灰度发布完成。

灰度发布的过程也叫作灰度放量，灰度放量是一种谨慎的产品运营手段。对于Android移动App产品而言，因为国内存在很多个应用下载市场，所以即使没有A/B测试系统，也可以利用应用市场实现灰度发布。即在发布产品新版本的时候，不是一次在所有应用市场同时发布，而是有选择地逐个市场发布。每发布一批市场，观察几天数据指标，如果没有问题，继续发布下一批市场。