美团点评 Hadoop/Spark 系统实践

2021-01-15 12:02 381人阅读评论(0)

系列文章

本文目录

系列文章
一、Hadoop/Spark 定位与应用架构
- 1.1 功能和定位（是什么）
- 1.2 在互联网企业的应用场景和外围系统（怎么用）
二、Hadoop / Spark 核心架构
三、美团点评架构改造案例

美团大数据系统整体架构

一、Hadoop/Spark 定位与应用架构

1.1 功能和定位（是什么）

Hadoop 开源地址和官网

Spark 开源地址和官网

简单来说，拆解之后的 Hadoop/Spark 项目基本分为三层：① 最底层是资源调度和文件存储 ② 往上是分布式计算内核 ③最上层是 Spark 衍生的一些框架。今天说的主要是三个黑框的部分。

我们可以看到，三个黑框部分有一个共同点——都有“分布式”。那为什么是分布式呢？
以下是一台单点机器的数据

这是美团对于数据存储和应用的需要：

我们可以看到，一块商用硬盘满打满算也就能写几十 T 的数据。但是美团每日单表的增量就有 235T 的数据。这是单节点机器远远无法满足的。这是不得已要做分布式的原因。

我们想要一个能够像个人电脑一样的目录树形式的去管理自己的文件，最好存储空间没有限制，操作简单，平台无关等等。但是现实却需要面临很多问题，节点挂掉、计算失败等等。那我们通过 Hadoop/Spark 这样的集群操作系统，通过访问统一的分布式操作接口，去实现从理想到现实的这样的一个过程。

这是HDFS暴露的接口，可以去访问管理集群的文件

同时提供了 Java 编程的接口。

Spark 也提供了同样类似的接口，支持 Java， Scala 和 Python。安装后从 bin/spark-shell 可以快速起一个 Spark 环境，同时封装了提交集群作业的接口。

在写代码的时候，右边的是分布式的 “Hello World”—— wordCount。无论数据量有多大，都可以统计出来每个字符的出现次数。左边一些特性，具体可以参考 API 文档。

YARN 资源调度

有不同的资源划分模式，可以划分给不同的组织结构，后面会详细讲到。

1.2 在互联网企业的应用场景和外围系统（怎么用）

直观的来看，主要分为以上三种应用，主要通过数据仓库支撑。下面分别举例讲述。

商业智能 —— 经营分析
一个旅游的漏斗模型，用于整个流量转化数据的建模和分析支撑。
搜索/推荐/广告系统——系统架构
各种数据的存储和处理和需要很强的分布式支持。
数据挖掘——单车停车点聚合
通过获得得各种基础数据，经过处理，把单车调度到大家可能要骑得地方。

上述三种应用场景都需要用到数据仓库的支持。我们常见的数据组织形式是 E-R 模型，即实体关系模型。但是在面对线上业务的时候，我们往往想要看到业务发生了哪些变更，或者需要恢复事实等，使用ER模型可能需要分表、大量连接，所以就产生新的数据组织形式——维度模型，面向分析性系统。

维度建模的过程简单来讲就是把数据拉到集群上，经过一系列的 sql 处理分别分成维度表、事实表等。

什么是事实呢？比如下面订单可能就是一个事实。整个表可能是一个门店的数据，最主要的是 poi_id。
通常关系型数据库是为了记录、修改和查询数据本身（增删改差），对数据本身所含有的业务含义并不关心。简单说是为了记录。而数据仓库则是重点关注数据与数据之间的业务含义，如不同统计口径与分析方式下数值差的意义。简单说是为了分析。那么为了保证分析结果的准确和有效，必须分解和细化分析口径，精确定位影响因素，维表和度量值就是定义分析角度和影响因素的一种方式。（参考：数据仓库为什么要用事实表和维度表）

这是美团点评大数据架构全景图

更加具体或者形象的来说，整个架构可以看作下图：

我们主要讲到的部分是最底层的一块，上层任务暂不细讲。（在说到成本优化的时候，老师说美团的利润非常薄，然后就不好意思的笑了）
下面是一些美团离线工具链关键模块：

我们回顾一下第一部分，讲了 Hadoop/Spark 的功能和定位（理想和现实），应用场景（三个例子），然后数据仓库的数据组织形式。

二、Hadoop / Spark 核心架构

2.1 核心架构 review

2.1.1 HDFS 分布式文件存储

Hadoop 简介

直观来看，Hadoop 架构如下：

这种配色的图呢，圆框代表一个进程，虚框代表一个物理机。虚线代表元数据的访问或者同步。

整体架构主要分成两部分，一个是 NameNode, 负责数据的组织管理，一个是 DataNode，负责数据的实际存储，DataNode 会通过心跳定时向 NameNode 汇报自己还活着。

Editslog ：保存了所有对 hdfs 中文件的操作信息
FSImage 是内存元数据在本地磁盘的映射，用于维护管理文件系统树，即元数据(metadata)。
hadoop 中 FsImage 与 Editslog 合并解析

然后集中精神，最难的一部分来了，就是数据的写流程。

图中的序号代表执行顺序，NN 是 NameNode, DN 是 DataNode，Lease 实际上是时间约束锁，对文件写操作的互斥同步靠Lease实现。

当你写入一个文件的时候到底发生了什么呢？

首先你打开的客户端（可能是我们之前提到的命令行或者连接了 SDK 的其他 IDE）会创建一个逻辑文件，建立心跳，然后客户端和NN手拉手建立管道， NameNode 找到空的 DN 分配给新文件并把分配的DN 的地址返回，然后客户端和第一个 DN 手拉手建立数据管道，第一个 DN（主节点）和其他 DNs（从节点，也叫副本，一般两个，防止数据丢失）手拉手建立数据管道。建好管道之后数据就通车了，把数据块先写进第一个DN，第一个DN写完之后写入第二个 DN，依次往下直至所有数据写入。最后一个 DN 写完之后会返回 ACK 确认码，说我终于写好了，接收到的 DN 继续传递（和烽火狼烟差不多）。客户端接收到确认信息之后就会关闭 Clint-DN 之间的管道。DN 给 NN 报告，我写完了。这样第一个数据块就写完了。
然后 NN 会继续找空的 DN，返回地址，……等等和以上过程一样，直到所有数据块全都写完。客户端和 NN 之间的管道关闭，数据写入结束。

如果写过程懂了，读过程就不用介绍了。如果没懂，那就再回去看看写过程（笑）。

如果在写入或者读取的过程，发生了意外情况，我们该怎么办呢？这就牵扯到容错处理了，但是这非常复杂，这里就简单介绍。

DN 挂掉一定时间以后，NN 会补上一个新的 DN。NN 挂掉会比较麻烦。为了不让 NN 挂掉产生可怕的影响，所以进行了一定的安全保障的设计。

作为一个小白，这里的设计术语也不是很明白，略过不讲。想要了解可以参考美团HDFS NameNode重启优化，或者参考社区解决方案。

以上两张图主要解决内存结点高可用，单节点扩展方法问题。

2.1.2 YARN 分布式资源调度

YARN 这里简单介绍。

YARN 这名字就像是硬凑的（老师说的，与我无瓜），其实就是资源协调器，为了使 Hadoop1.0 版本中资源调度和计算框架进行分析而提出的。YARN 的模块设计如下，主要包含三个 Manager。

任务启动流程可大致了解。

资源调度策略——FairScheduler（公平调度）

这是 FairScheduler 的内部实现。

具体有哪些资源调度以及什么原理我觉得可以参考这个：Hadoop 的三种调度器

2.1.3 Spark 通用分布式计算框架

Spark 自称比 Hadoop 有100倍提升。

除了支持多语言，也提供了四个框架，包括 Spark SQL，Spark Streaming（流式框架），MLlib（机器学习库），GraphX（图计算库）等，比较完备。

Spark 具体暴露了什么样的编程接口呢？下图还是上面提到的 “Hello World” 。通过一个入口命令行启动自己的程序，自己声明一个 Spark Session 或者命令行提供的 Session 去构造 RDD（弹性分布式数据集）的一个最底层的逻辑抽象。可以把 RDD看作一个大的 list 来用，里面包含各种文件和数据。RDD 包含两个算子，一个叫做 Transform，理解为数据的转换操作；一个叫 Action，理解为对数据的计算操作。

这里是一些其他接口：

这里举了一个 wordcount 的例子。右边的图使左边代码的一个直观体现，主要对应了左边红框部分的运算。

直观来看，flatMap，map，reduceByKey 是一个 transform 因为其将语料进行了切分，就是进行了数据的转换。而 saveAsFile 都是 Action，因为其对数据进行计算了存储。（个人理解）

我们写好了程序，又该如何提交呢？我们来看一下 Spark 的提交流程，其实和我们的 YARN 的架构很相似。Driver 相当于 YARN 的 app master，基于自己的调度逻辑和 RM 拿资源，拿到资源后把 Spark Executer 提起来。

这是一个 RDD 的一个描述图，最左边可能是一个 Java 程序，然后通过逻辑进行分割成不同的 stages，每一个小框（一个逻辑）包含几个块（task）。对于每个task，要到集群（Cluster manager ）上申请资源，然后调度到实际在跑的结点上去执行该执行的逻辑。这里的 Worker 就是 Executor。

这些概念有印象就行，一般也只有 debug 才会用到。

前面我们提到了 Stage，那么 Satge 具体怎么拆分的呢？
用户提交的计算任务是一个由 RDD 构成的 DAG，如果 RDD 在转换的时候需要做 Shuffle，那么这个 Shuffle 的过程就将这个 DAG 分为了不同的阶段（即Stage）。由于 Shuffle 的存在，不同的Stage 是不能并行计算的，因为后面 Stage 的计算需要前面 Stage 的 Shuffle 的结果。

然后我们来看一个比较完整的包含 shuffle 的 mapreduce 的例子。（大家注意区分 mapredcue 中的 map 和 map func 中的 map 的区别）

*蓝色是数据流转的过程，圆角黑框是进程，Memory 部分是内存部分

流程如下：

Map 端通过 HDFS Client 读入数据，执行 Map 端用户自定义逻辑。
通过 key 和用户定义 Map Func 对数据进行分区，中间结果写入缓存。
缓存空间不够时（一般使用到80%）会将缓存数据写入文件，保存到磁盘，这个过程叫 Spill（这里输出的数据是根据 key 组织，相对有序的）。然后就可以继续写入内存，直到所有数据被处理。
对于所有 Spill 得到的文件（文件之间无序）和内存数据，进行 Merge Sort 按序写入大文件。一个 Map 端的结果可能被不同的 Reduce 读取，所以会有一个 Map 和 Reduce 的映射关系，以及文件是怎么切分的都会记录在Spark 中的 MapOutput 元数据中。
所有 Map 端 task 跑完之后，Reduce 端服务启动，拉取 Shuffle 之后的 Map 端结果。
拉取数据先写缓存。
缓存不够 Spill 到本地磁盘。
合并所有 Spill 文件。
执行 Reduce 端用户定义逻辑
按需输出数据，写HDFS。

总的来说：
Map 端：map -> partition -> spill -> merge sort -> shuffle
Reduce 端：fetch -> spill -> merge sort -> reduce -> some where

Spark 的容错相对简单，一个 Worker 挂了，就把对应 Task 调度到其他 Worker 就行了。如果 Driver 结点挂掉了，YARN 会帮忙拉起来。