互联网技术的发展让大多数企业能够积累大量的数据，而企业需要灵活快速地从这些数据中提取出有价值的信息来服务用户或帮助企业自身决策。然而处理器的主频和散热遇到了瓶颈，CPU难以通过纵向优化来提升性能，所以多核这种横向扩展成为了主流。也因此，开发者需要利用多核甚至分布式架构技术来提高企业的大数据处理能力。这些技术随着开源软件的成功而在业界得到广泛应用。
下面我稍微介绍一些大数据应用中通常出现的一些原理或者说是特征。
基本原理
分布式：将数据分布到不同的节点（机器），从而存储大量的数据。而分布式同时为并行读写和计算提供了基础，这样就能提高数据的处理能力。
为什么不直接使用分布式的关系型数据库，比如主从模式的mysql？这主要是效率的问题。分布式关系型数据库为了实现分布式事务、线性一致性、维护自身索引结构等功能会对性能造成影响。而正如刚刚背景所提到，大数据需求重点是快速处理大量数据，帮助用户和企业的决策。这个决策就包括推荐、监控、数据分析等。这些场景并不一定需要数据库这种严格的约束，是OLAP而非OLTP。所以大数据技术会通过解除这些限制而提升性能。
除了分布式外，还可以利用
可靠性相关
由于这个分享会的标题起得有点大，包括存储、搜索、计算三大块，而且篇幅有限，所以我就只根据这三块中我了解且比较流行的开源组件来分享，而且只讲解大概的原理。毕竟下面的每个组件的原理和实战都可以各自出一本五六百页的书了，而且还没涉及源码细节的。下面首先来介绍分布式文件系统，就是把我们单台计算机的文件系统扩展到多台。
HDFS（Hadoop Distributed File System）
架构原理
图中有8台机器或者容器，两个client、5个DataNode、1个NameNode。一个分布式文本系统，组成：NameNode、DataNode和secondary namenode
读写流程
写入：client端调用filesystem的create方法，后者通过RPC调用NN的create方法，在其namespace中创建新的文件。NN会检查该文件是否存在、client的权限等。成功时返回FSDataOutputStream对象。client对该对象调用write方法，这个对象会选出合适存储数据副本的一组datanode，并以此请求DN分配新的block。这组DN会建立一个管线，例如从client node到最近的DN_1，DN_1传递自己接收的数据包给DN_2。DFSOutputStream自己还有一个确认队列。当所有的DN确认写入完成后，client关闭输出流，然后告诉NN写入完成。
读取：client端通过DistributedFileSystem对象调用open方法，同样通过RPC调用远程的NN方法获取所要查询的文件所涉及的blocks所存储的DN位置，而且这些位置是按照距离排序的。返回的结果是一个FSDataInputStream对象，对输入流对象调用read方法。输入流会从距离最近的DN中读取数据，将数据传递到client，读取结束后关闭流。
这个机制看上去是很笨重的，有了这个分布式文件系统的基础，其他组件就能利用这个系统提供的 API 来对数据的存储进行优化。在介绍下一个组件前，先对主要的主键索引作简单的介绍。
索引
类型
哈希SSTables/LSM树BTree/B+Tree大致原理数据结构：哈希表。内存：有序集合，例如红黑树、平衡二叉树、跳跃表。
磁盘：一个个独立文件，里面包含一个个数据块。
写入：内存维护一个有序集合，数据大小达到一定阈值写入磁盘。后台会按照特定策略合并segment。读取：先查询内存，然后磁盘中的最新segment，然后第二新，以此类推。数据结构：平衡多叉树。写入：通过二分查找找到相应的叶子结点进行修改。读取：同上。优势适合数据经常更新写入快，顺序读取快，容易压缩读取快，更时间可控劣势必须存储在内存；范围查询效率低随机读取，读取旧数据较慢写入较慢涉及数据库Mysql、RedisMongoDB、Elasticsearch、HBaseMysql、MongoDB
主要的主键索引有哈希、LSM、BTree。下面主要涉及到LSM树，所以哈希和BTree这里就不多说了。LSM树有内存和磁盘两个部分....，以跳跃表为例，大致的模型如下图
内存的 MemStore 是一个有序集合，数据写入会先写入这里，当大小达到阈值就会 flush 到磁盘。而后台会有程序按一定策略对这些文件进行合并。合并的原因有：减少小文件，进而减少读取时IO来提升读性能。数据合并，比如图中第二个file有数据a，但现在客户端发送请求要把它删掉或进行修改，如果每次删改都要把数据找到再调整，就会有大量的磁盘IO，所以这些操作一般只做标记，等到后续文件合并时才真正对数据进行修改。还有一个原因是调整排序，因为flush后数据只在file内部有序，合并能够调整整体排序。正因为这种结构，所以LSM的写入是很快的，范围读取也快，因为数据已经有序。而为了保证不读取到旧版本的数据，所以读取需要从最新的开始遍历，这也导致读取旧数据的效率较低。当然，这里面还能优化，但细节就不说了。
HBase
简介
HBase 就是基于 HDFS API 构建的一个可以在线低延迟访问大数据的NoSQL数据库。本质上就是给 HDFS 加上一个 LSM Tree 索引，从而提高读写性能。当然，即便优化了，这个高性能也是相对大数据量而言。实际上“HBase并不快，只是当数据量很大的时候它慢的不明显”。由于是 NoSQL 数据库，所以它有文档型数据库的弱项，即基本不支持表关联。
特点
- 数据量大，单表至少超千万。对稀疏数据尤其适用，因为文档型数据库的 null 就相当于整个字段没有，是不需要占用空间的。
- 高并发写入，正如上面 LSM 树所说。
- 读取近期小范围数据，效率较高，大范围需要计算引擎支持。
- 数据多版本
- 小数据
- 复杂数据分析，比如关联、聚合等，仅支持过滤
- 不支持全局跨行事务，仅支持单行事务
场景
更多适用场景可以根据HBase的特点判断
架构原理
这里大概有10台机器或节点，5个DataNode、两个RegionServer、一个Client、Master、ZooKeeper
- 实现Master的高可用：当active master宕机，会通过选举机制选取出新master。
- 管理系统元数据：比如正常工作的RegionServer列表。
- 辅助RS的宕处理：发现宕机，通知master处理。
- 分布式锁：方式多个client对同一张表进行表结构修改而产生冲突。
- 处理 client 的 DDL 请求
- RegionServer 数据的负载均衡、宕机恢复等
- 清理过期日志
- Store：一个Store存储一个列簇，即一组列。
- MemStore和HFile：写缓存，阈值为128M，达到阈值会flush成HFile文件。后台有程序对这些HFile进行合并。
- HLog（WAL）：提高数据可靠性。写入数据时先按顺序写入HLog，然后异步刷新落盘。这样即便 MemoStore 的数据丢失，也能通过HLog恢复。而HBase数据的主从复制也是通过HLog回放实现的。
- BlockCache
- Region：数据表的一个分片，当数据表大小达到一定阈值后会“水平切分”成多个Region，通常同一张表的Region会分不到不同的机器上。
读写过程
写入和读取的流程类似。
ElasticSearch
简介
Elastic Stack 是以 Elasticsearch 为中心开发的一组组件，其中Kibana、Logstash、Beats使用较多。
Beats 是用 GO 实现的一个开源的用来构建轻量级数据汇集组件，可用于将各种类型的数据发送至 Elasticsearch 与 Logstash。
Logstash：流入、流出 Elasticsearch 的传送带。其他MQ或计算引擎也可以导入ES。
利用 Logstash 同步 Mysql 数据时并非使用 binlog，而且不支持同步删除操作。
Kibana 是 ES 大数据的图形化展示工具。集成了 DSL 命令行查看、数据处理插件、继承了 x-pack（收费）安全管理插件等。
Elasticsearch 搜索引擎，它并不是基于 HDFS 建立的，而是自己实现了分布式存储，并通过各种索引和压缩技术来提高搜索的性能。当然，它作为文档型数据库，其在内存组织数据的方式也是类似LSM树的。
特点
- 全文检索，like "%word%"
- 一定写入延迟的高效查询
- 多维度数据分析
- 关联
- 数据频繁更新
- 不支持全局跨行事务，仅支持单行事务
场景
框架原理
Cluster
Node：JVM进程
Index：一组形成逻辑数据存储的分片的集合，数据库
Shard：Lucene 索引，用于存储和处理 Elasticsearch 索引的一部分。
Segment：Lucene 段，存储了 Lucene 索引的一部分且不可变。结构为倒排索引。
Document：条记录，用以写入 Elasticsearch 索引并从中检索数据。
增删改查原理
Update = Delete + (Index - Ingest Pipeline）
细节补充
倒排索引
一般正向的就是通过文档id找相应的值，而倒排索引则是通过值找文档id。通过倒排这种结构，判断哪些文档包含某个关键词时，就不需要扫描所有文档里面的值，而是从这个关键词列表中去搜索即可。而频率主要是用来计算匹配程度的，默认使用TF-IDF算法。
为什么全文检索中 ES 比 Mysql 快？
Mysql 的辅助索引对于只有一个单词的字段，查询效率就跟 ES 差距不大。
select field1, field2
from tbl1
where field2 = a
and field3 in (1,2,3,4)
这里如果 field2 和 field3 都建立了索引，理论上速度跟 es 差不多。es最多把 field2 和 field3 concat 起来，做到查询时只走一次索引来提高查询效率。
但如果该字段是有多个单词，那么缺乏分词的 Mysql 就无法建立有效的索引，且查询局限于右模糊，对于“%word%”的搜索效率是极低的。而 ES 通过分词，仍然可以构建出 term dictionary。
然而 Term Dictionary 和 Position 加起来是很大的，难以完全存储在内存。因此，在查找 Term Dictionary 的过程会涉及磁盘IO，效率就会降低。为此，Luence 增加了 term index。这一层通过 Lucene 压缩算法，使得整个 Term Index 存储在内存成为可能。搜索时在内存找到相应的节点，然后再到 Term Dictionary 找即可，省去大量磁盘IO。
内存消耗大
ES 之所以快，很大程度是依赖 Lucene 的缓存以及缓存中的索引结构。而这些缓存只有被预先加载到内存才能做到快速的响应，查询没有被加载的数据通常都是比较慢的，这是 ES 需要大量内存的原因之一。所以有人建议 ES 仅作为内存索引库，即与where、group by、in、sort等过滤、聚合相关的才存储到 ES，而且其他字段并不能帮助查询，只会浪费内存空间。而查询得出的id将返回通过 Mysql 或者 Hbase 进行第二次的查询。由于是主键的搜索，所以不会耗费太多时间。
而 ES 由于给了大部分内存到 Lucene 缓存，那自己聚合计算时用的内存空间就很有限了，这也是 ES 需要大量内存的原因。
目前触漫 ES 情况
刚刚起步，仅仅作为优化部分慢sql查询的解决方案。而 ES 更强大的准实时数据分析、文本搜索功能并没有开发。这其中有涉及到搜索优化（排序规则、分词等）、Kibana可视化、数据冷热分离、各种配置等，所以是需要一定的人力去学习和调试才能发挥它的潜能。
从上面的介绍我们可以知道，ES 是不支持关联的，而且聚合计算的资源很有限。那这时就用到计算引擎了。
计算引擎
计算引擎目前主流的两个开源组件分别是 Spark 和 Flink。从两个引擎的处理模型来看，Spark 的批处理更为高效，Flink 则善于流处理，尽管两者都向着流批一体化的方向发展。当然，只要对弱项做优化还是可以跟另一方未做太多优化的强项比的，只是实现难度大些和效果上限可能略低。比如 Blink，阿里内部的 Flink，其 ML 模块经过优化，在大部分常用模型的计算效率都能高于开源的 Spark 的。如果开源 Spark 也经过阿里那样深度的优化，两者的差距就难说了。
简单提一下他们的特点
Spark
下面首先介绍 Spark，它是一个用于大规模数据处理的统一分析引擎，其内部主要由 Scala 实现。Spark 当初引起关注主要是它与 Hadoop 的三大件之一的 MapReduce 之间的比较。Hadoop 的三大组件包括 HDFS、Yarn 和 MapReduce。他们三个都是可以拆分开来单独使用的。比如 Yarn 作为资源调度系统，传统 Spark 和 Flink 都会借助它的功能实现任务的调度。而 MapReduce 作为计算引擎，其计算速度当时是弱于 Spark 的，主要是 Spark 减少了不必要的磁盘IO；增加迭代计算功能，从而更好支持机器学习；引入了一些自动优化功能。另外，Spark 广泛的语言支持、API 更强的表达能力等优点都让 Spark 在当时的离线计算领域中超越 MapReduce。
功能丰富
4大场景：Spark 的高层组件包括Spark SQL、Spark Streaming、Spark ML、GraphX。他们都是通过底层组件为 Spark Core 实现具体功能的。但是在使用 Spark 的时候，尽量是不要使用 Spark Core，因为高层组件的产生的 Spark Core一般会更高效，因为Spark做了不少优化，具体后面再说。
多种语言：支持 Java、Python、R 和 Scala 来编写应用代码。
多种部署模式：本地、独立部署、Mesos、Yarn、K8S
多种数据源：HDFS、HBase、Hive、Cassandra、Kafka等
架构原理
Driver 是启动 Spark 作业的JVM进程，它会运行作业（Application）里的main函数，并创建 SparkContext 对象。这个 SparkContext 里面包含这次 Spark 计算的各种配置信息。Spark 通过它实现与 Cluster Manager 通信来申请计算资源。这里的 Cluster Manager，在生产环境一般是 Mesos、Yarn 或者 K8s。这些 Manager 根据其管理的集群情况，给这个 Spark 任务分配相应的容器container，在容器中启动 executor 进程。这些启动后的 executor 会向 Driver 注册，之后 Driver 就可以把它根据用户计算代码生成出的计算任务task发送给这些 executor 执行。计算结束后，结果可能输出到 Driver，也可能输出到当前 executor 的磁盘，或者其他存储。
作业例子
object SparkSQLExample {def main(args: Array[String]): Unit = {// 创建 SparkSession，里面包含 sparkcontextval spark = SparkSession.builder().appName("Spark SQL basic example").getOrCreate()import spark.implicits._// 读取数据val df1 = spark.read.load("path1...")val df2 = spark.read.load("path2...")// 注册表df1.createOrReplaceTempView("tb1")df2.createOrReplaceTempView("tb2")// sqlval joinedDF = sql("""|select tb1.id, tb2.field|from tb1 inner join tb2|on tb1.id = tb2.id""".stripMargin)// driver 终端显示结果joinedDF.show()// 退出 sparkspark.stop()}}
SQL会经过一层层的解析然后生成对应的 Java 代码来执行。
计算引擎的优势
与 HBase、 es 和传统数据库查询比较，计算引擎的优势：1）数据量大时速度快，2）计算更加灵活。
以大数据关联为例：
- 基于内存：计算引擎留有大量内存空间专门用于计算，尽量减少磁盘 IO。
- 计算并行化
- 算法优化
具体而言，Spark 提供了三种 Join 执行策略：
从上面的例子可以看出计算引擎相比于其他组件在计算方面的优势。
数据流动
下面通过一张图，从另一个角度了解 Spark 的运作。
这是一张简单的数据流程图。描述了一个 WorkCount 的数据流向。其主要代码如下：
// 假设每个 block 里的数据如下// a// b// aval textFile = sc.textFile("hdfs://...")val counts = textFile.map(word => (word, 1)) // a -> <a,1>.reduceByKey(_ + _) // <a,<1,1>> -> <a,2>counts.saveAsTextFile("hdfs://...")
图中同一阶段有多个数据流体现的是并行。中间的 shuffle 是在聚合、关联、全局排序等操作时会出现的。比如这里的 reduceByKey 就是将相同 key 的数据移动到相同的 partition。这样就能对所有的 a 进行加总，从而得出 a 的总数。
上图的任务是一次性的，或者是周期性的，数据的驱动是拉取型的。如果将数据块换成数据流，map 和 reduce 在启动后就一直存在，并接受数据源不断发送过来的信息，那就变成了流计算。即由周期性变为一直处理，从而变为实时处理，由主动拉取变为被动接收的形式。下面就来介绍 Flink 计算引擎。
Flink
Flink 同样是分布式的计算引擎，主要基于Java实现，但它的特色主要体现在流式计算。这个引擎流行的主要推手是阿里。阿里在19年初开源了它修改过的 Flink，收购了 Flink 的母公司，并在各种线下技术论坛上推广 Flink，让 Flink 在 19 年的关注度极速上升。
除了在实时计算领域，Flink 在其他领域或许稍微落后于 Spark，毕竟 Spark 发展比较早，其生态比 Flink 要成熟更多。Flink 目前支持 Scala、Java 和 Python 来写任务代码。功能上同样支持批计算、ML、Graph。部署工具、支持的数据源也 Spark 类似。
场景
架构原理
细节补充
和 Spark 一样，Flink 也会根据 SQL 或者业务代码生成 DAG 图，然后将任务划分并发送给不同的节点执行。最大的不同正如之前所说，数据是实时地、一条条或一小批一小批地不断流进这些节点，然后节点输出响应的结果。而在这种场景下，Flink 在一定程度上解决了实时处理中的不少难点。
与 Spark 比较
Spark：
Flink：
小红书实时技术
小红书旧的离线框架和我们现在的大数据体系有点类似，都是把埋点数据上报到日志服务，然后进入离线数仓，只是小红书用 Hive，我们用 DataWorks。然后我们同样也有 T+1 的用户画像、BI报表和推荐的训练数据。
而后续的实时框架是这样的
日志服务的埋点数据先进入 Kafka 这一消息队列里面。不太清楚为什么要加上 Kafka 这一中间件，或许当时并没有开源的 日志服务到Flink 的 connecter 吧。但总之，引入 Flink 之后就可以实时累计埋点中的数据，进而产生实时的画像、BI指标和训练数据了。下面介绍一下这个实时归因
如上图所以，用户app屏幕展示了4个笔记，然后就会有4条曝光埋点，而如果点击笔记、点赞笔记以及从笔记中退出都会有相应的埋点。通过这些埋点就可以得出右面两份简单的训练或分析数据。这些数据跟原来已经积累的笔记/用户画像进行关联就能得出一份维度更多的数据，用于实时的分析或模型预测。实时模型训练这一块至少小红书在19年8月都还没有实现。下图是小红书推荐预测模型的演进
那么如何进行实时训练深度学习模型呢？以下是我的一些想法。借助一个阿里的开源框架flink-ai-extended。
如上图所示，这是 flink 的数据流结构图，左边 source 为数据源，然后进过join、udf等算子进行训练样本数据的生成，然后传递给一个 UDTF/FlatMap 算子，这实际上也是一个 Flink 节点，但它里面包含的是 Tensorflow 的训练 worker，而上下也是 Flink 的节点，都是包含了 Tensorflow 训练所需的一些角色，这样数据源源不断地实时进入 TF 模型来完成实时训练。TF 也可以因此借助 Flink 的分布式框架来完成分布式的学习。多台GPU或者CPU或许应该会比一台GPU的训练效率更高。
这个框架同时适用于模型预测，只要把里面的训练角色换成训练完成的 model，也就可以进行实时的预测，而且这里借助 Flink 内部的通信机制，效率应该会比普通的 http 调用要快不少。
总结
本次分享由于时间有限，讲的都是比较浅层的东西，实际上刚刚所说的每一个组件里面包含的内容都不少，都可以作为一个长远的目标去研究和改造。说回分享的主题之一，使用场景。
首先是存储，上述介绍的 HDFS、HBase、ES（ES虽然是搜索引擎，但它也可以在某些方面替代传统关系型数据的功能） 都是适用于 OLAP 场景，即分析推荐而非事务。从公司目前的情况来看，HDFS 基本可以忽略，因为已经有 DataWork，数据的存储暂时不是问题。更多的问题在于数据使用时的性能。HBase 和 ES 作为文档型数据库，适合一对多的数据模型，比如将帖子和其评论作为一个整体来存储。对于多对一、多对多的模型，文档型数据库实际上并不合适，但可以通过合并宽表、应用层关联等方式在一定程度上进行弥补。而如果多对多关系确实复杂、量大、文档型数据库性能无法满足，比如一些大型社交网络，那么可以考虑图数据库。
当决定尝试文档型数据库时，HBase 的特点在于较为快速地查询小范围的新数据，而且这条数据可以很大。ES 的特点则在于快速的全文检索、准实时的数据分析。当然，分析的复杂度是不能跟计算引擎比的，比如关联、机器学习等。但通过合并宽表、各种where、group by操作，还是能满足不少需求的，尤其是应用的搜索功能，ES 实现起来是比较简单的。目前公司并没有应用它的强项，最好由专人负责它的调试，尤其是搜索排序方面。
然后是计算引擎，目前公司用的 MaxCompute 已经能够满足离线计算的各种需求，或者就欠缺实时计算了。但公司目前实时性需求不多而且也不紧急，所以开发一直都没有启动。目前就看明年推荐是否有这样的需求，而且有相应的prd出来了。而考虑到成本和灵活性，自建或许是更好的选择，比如刚刚提到的 Flink + Tensorflow。
以上便是这次分享会的全部内容，谢谢大家的参与。
参考：
书籍：
文章：
文档：
分享：
