阿里云云原生数据湖体系全解读——数据湖云原生计算引擎

云原生背景介绍与思考

图是基于 ECS 底座的 EMR 架构，这是一套非常完整的开源大数据生态，也是近
10 年来每个数字化企业必不可少的开源大数据解决方案。主要分为以下几层：
ECS 物理资源层，也就是 Iaas 层。数据接入层，例如实时的 Kafka，离线的 Sqoop。存储层，包括 HDFS 和 OSS，以及 EMR 自研的缓存加速 JindoFS。计算引擎层，包括熟知的 Spark， Presto、 Flink 等这些计算引擎。数据应用层，如阿里自研的 DataWorks、 PAI 以及开源的 Zeppelin， Jupyter。

每一层都有比较多的开源组件与之对应，这些层级组成了最经典的大数据解决方案，也
就是 EMR 的架构。我们对此有以下思考：是否能够做到更好用的弹性，也就是客户可以完全按照自己业务实际的峰值和低谷进行弹性扩容和缩容，保证速度足够快，资源足够充足。不考虑现有状况，看未来几年的发展方向，是否还需要支持所有的计算引擎和存储引擎。这个问题也非常实际，一方面是客户是否有能力维护这么多的引擎，另一方面是是否某些场景下用一种通用的引擎即可解决所有问题。举个例子说 Hive 和 Mapreduce，诚然现有的一些客户还在用 Hive on Mapreduce，而且规模也确实不小，但是未来Spark 会是一个很好的替代品。存储与计算分离架构，这是公认的未来大方向，存算分离提供了独立的扩展性，客户可以做到数据入湖，计算引擎按需扩容，这样的解耦方式会得到更高的性价比。

基于以上这些思考，我们考虑一下云原生的这个概念，云原生比较有前景的实现就是Kubernetes，所以有时候我们一提到云原生，几乎就等价于是 Kubernetes。随着Kubernetes 的概念越来越火，客户也对该技术充满了兴趣，很多客户已经把在线的业务搬到了 Kubernetes 之上。并且希望在这种类似操作系统上，建设一套统一的、完整的大数据基础架构。所以我们总结为以下几个特征：

希望能够基于 Kubernetes 来包容在线服务、大数据、 AI 等基础架构，做到运维体系统一化。
存算分离架构，这个是大数据引擎可以在 Kubernetes 部署的前提，未来的趋势也都在向这个方向走。
通过 Kubernetes 的天生隔离特性，更好的实现离线与在线混部，达到降本增效目的。
Kubernetes 生态提供了非常丰富的工具，我们可以省去很多时间搞基础运维工作，从而可以专心来做业务。

EMR 计算引擎 on ACK

ACK 就是阿里云版本的 Kubernetes，在兼容社区版本的 API 同时，在本文中不会区分 ACK 和 Kubernetes 这两个词，可以认为代表同一个概念。
基于最开始的讨论，我们认为比较有希望的大数据批处理引擎是 Spark 和 Presto，当然我们也会随着版本迭代逐步的加入一些比较有前景的引擎。EMR 计算引擎提供以 Kubernetes 为底座的产品形态，本质上来说是基于CRD+Operator 的组合，这也是云原生最基本的哲学。我们针对组件进行分类，分为service 组件和批处理组件，比如 Hive Metastore 就是 service 组件， Spark 就是批处理组件。
图中绿色部分是各种 Operator，技术层面在开源的基础上进行了多方面的改进，产品层面针对 ACK 底座进行了各方面的兼容，能够保证用户在集群中很方便的进行管控操作。右边的部分，包括 Log、监控、数据开发、 ECM 管控等组件，这里主要是集成了阿里云的一些基础设施。我们再来看下边的部分：
 引入了 JindoFS 作为 OSS 缓存加速层，做计算与存储分离的架构。
 打通了现有 EMR 集群的 HDFS，方便客户利用已有的 EMR 集群数据。
 引入 Shuffle Service 来做 Shuffle 数据的解耦，这也是 EMR 容器化区别于开源方案的比较大的亮点，之后会重点讲到。

Spark on Kubernetes 的挑战

整体看， Spark on Kubernetes 面临以下问题：
我个人认为最重要的，就是 Shuffle 的流程，按照目前的 Shuffle 方式，我们是没办法打开动态资源特性的。而且还需要挂载云盘，云盘面临着 Shuffle 数据量的问题，挂的比较大会很浪费，挂的比较小又支持不了 Shuffle Heavy 的任务。
调度和队列管理问题，调度性能的衡量指标是，要确保当大量作业同时启动时，不应该有性能瓶颈。作业队列这一概念对于大数据领域的同学应该非常熟悉，他提供了一种管理资源的视图，有助于我们在队列之间控制资源和共享资源。读写数据湖相比较 HDFS，在大量的 Rename， List 等场景下性能会有所下降，同时OSS 带宽也是一个不可避免的问题。

Spark on Kubernetes 的解决方案

Remote Shuffle Service 架构
Spark Shuffle 的问题，我们设计了 Shuffle 读写分离架构，称为 RemoteShuffle Service。首先探讨一下为什么 Kubernetes 不希望挂盘呢，我们看一下可能的选项：
 如果用是 Docker 的文件系统，问题是显而易见的，因为性能慢不说，容量也是极其有限，对于 Shuffle 过程是十分不友好的。
 如果用 Hostpath，熟悉 Spark 的同学应该知道，是不能够启动动态资源特性的，这个对于 Spark 资源是一个很大的浪费，而且如果考虑到后续迁移到 Serverless K8s，那么从架构上本身就是不支持 Hostpath 的。
 如果是 Executor 挂云盘的 PV，同样道理，也是不支持动态资源的，而且需要提前知道每个 Executor 的 Shuffle 数据量，挂的大比较浪费空间，挂的小 Shuffle 数据又不一定能够容纳下。
所以 Remote Shuffle 架构针对这一痛点、对现有的 Shuffle 机制有比较大的优化，图 3 中间有非常多的控制流，我们不做具体的讨论，具体架构详见文章《 ServerlessSpark 的弹性利器 - EMR Shuffle Service》。主要来看数据流，对于 Executor 所有的 Mapper 和 Reducer，也就是图中的蓝色部分是跑在了 K8s 容器里，中间的架构是Remote Shuffle Service，蓝色部分的 Mapper 将 Shuffle 数据远程写入 service 里边，消除了 Executor 的 Task 对于本地磁盘的依赖。 Shuffle Service 会对来自不同Mapper 的同一 partition 的数据进行 merge 操作，然后写入到分布式文件系统中。等到Reduce 阶段， Reducer 通过读取顺序的文件，可以很好的提升性能。这套系统最主要的实现难点就是控制流的设计，还有各方面的容错，数据去重，元数据管理等等工作。
简而言之，我们总结为以下 3 点：
 Shuffle 数据通过网络写出，中间数据计算与存储分离架构
 DFS 2 副本，消除 Fetch Failed 引起的重算， Shuffle Heavy 作业更加稳定
 Reduce 阶段顺序读磁盘，避免现有版本的随机 IO，大幅提升性能

Remote Shuffle Service 性能
我们在这里展示一下关于性能的成绩，图 4 是 Terasort 的 Benchmark 成绩。之所以选取 Terasrot 这种 workload 来测试，是因为他只有 3 个 stage，而且是一个大Shuffle 的任务，大家可以非常有体感的看出关于 Shuffle 性能的变化。左边图，蓝色是Shuffle Service 版本的运行时间，橙色的是原版 Shuffle 的运行时间。我们观察有 2T，4T， 10T 的数据，可以看到随着数据量越来越大， Shuffle Service 的优势就越明显。右图观察，作业的性能提升主要体现在了 Reduce 阶段，可以看到 10T 的 Reduce Read从 1.6 小时下降到了 1 小时。原因前边已经解释的很清楚了，熟悉 spark shuffle 机制的同学知道，原版的 sort shuffle 是 M*N 次的随机 IO，在这个例子中， M 是 12000， N 是8000，而 Remote Shuffle 就只有 N 次顺序 IO，这个例子中是 8000 次，所以这是提升性能的根本所在。

其他方面的重点优化
这里讲一下 EMR 在其他方面做得优化
调度性能优化，我们解决了开源的 Sp a r k Op e ra t o r 的一些不足，对于Executor pod 的很多配置 Spark Operator 把他做到了 Webhook 里边，这个对调度来说是十分不友好的，因为相当于在 API Server 上绕了一圈，实际测试下来性能损耗很大。我们把这些工作直接做到 spark 内核里边，这样避免了这方面的调度性能瓶颈。然后从调度器层面上，我们保留了两种选择给客户，包括 ACK 提供的
SchedulerFrameworkV2 实现方式和 Yunicorn 实现方式。读写 OSS 性能优化，我们这里引入了 JindoFS 作为缓存，解决带宽问题，同时
EMR 为 OSS 场景提供了 Jindo Job Committer，专门优化了 job commit 的过程，大大减少了 Rename 和 List 等耗时操作。针对 Spark 本身， EMR 积累了多年的技术优势，也在 TPCDS 官方测试中，取得了很好的成绩，包括 Spark 性能、稳定性，还有 Delta lake 的优化也都有集成进来。我们提供了一站式的管控，包括 Notebook 作业开发，监控日志报警等服务，还继承了 NameSpace 的 ResourceQuota 等等。

Spark 云原生后续展望

从我的视角来观察， Spark 云原生容器化后续的方向，一方面是达到运维一体化，另一方面主要希望做到更好的性价比。我们总结为以下几点：
 可以将 Kubernetes 计算资源分为固定集群和 Serverless 集群的混合架构，固定集群主要是一些包年包月、资源使用率很高的集群， Serverless 是做到按需弹性。
 可以通过调度算法，灵活的把一些 SLA 不高的任务调度到 Spot 实例上，就是支持抢占式 ECI 容器，这样可以进一步降低成本。
 由于提供了 Remote Shuffle Service 集群，充分让 Spark 在架构上解耦本地盘，只要 Executor 空闲就可以销毁。配合上 OSS 提供的存算分离，必定是后续的主流方向。
 对于调度能力，这方面需要特别的增强，做到能够让客户感受不到性能瓶颈，短时间内调度起来大量的作业。同时对于作业队列管理方面，希望做到更强的资源控制和资源共享。

EMR Shuffle Service

计算存储分离是云原生的重要特征。通常来讲，计算是 CPU 密集型，存储是 IO 密集型，他们对于硬件配置的需求是不同的。在传统计算存储混合的架构中，为了兼顾计算和存储， CPU 和存储设备都不能太差，因此牺牲了灵活性，提高了成本。在计算存储分离架构中，可以独立配置计算机型和存储机型，具有极大的灵活性，从而降低成本。
存储计算分离是新型的硬件架构，但以往的系统是基于混合架构设计的，必须进行改造才能充分利用分离架构的优势，甚至不改造的话会报错，例如很多系统假设本地盘足够大，而计算节点本地盘很小；再例如有些系统在 Locality 上的优化在分离架构下不再适用。Spark Shuffle 就是一个典型例子。

每个 mapper 把全量 shuffle 数据按照 partitionId 排序后写本地文件，同时保存索引文件记录每个 partition 的 offset 和 length。reduce task 去所有的 map 节点拉取属于自己的 shuffle 数据。大数据场景 T 级别的 shuffle 数据量很常见，这就要求本地磁盘足够大，导致了跟计算存储分离架构的冲突。因此，需要重构传统的 shuffle 过程，把 shuffle 数据卸载到存储节点。

除了计算存储分离架构下的刚需，在传统的混合架构下，目前的 shuffle 实现也存在重要缺陷:大量的随机读写和小数据量的网络传输。考虑 1000 mapper * 2000 reducer的 stage，每个 mapper 写 128M shuffle 数据，则属于每个 reduce 的数据量约为 64k。从 mapper 的磁盘角度，每次磁盘 IO 请求随机读 64K 的数据; 从网络的角度，每次网络请求传输 64k 的数据：都是非常糟糕的 pattern，导致大量不稳定和性能问题。因此，即使在混合架构下，重构shuffle 也是很必要的工作。

EMR Shuffle Service设计

基于以上的动机，阿里云 EMR 团队设计开发了 EMR Shuffle Service 服务(以下称ESS)，同时解决了计算存储分离和混合架构下的 shuffle 稳定性和性能问题。
ESS 包含三个主要角色: Master, Worker, Client。其中 Master 和 Worker 构成服
务端， Client 以不侵入方式集成到 Spark 里。 Master 的主要职责是资源分配和状态管理；Worker 的主要职责是处理和存储 shuffle 数据； Client 的主要职责是缓存和推送 shuffle数据。整体流程如下所示(其中 ResourceManager 和 MetaService 是Master 的组件)：

ESS 采用 Push Style 的 shuffle 模式，每个 Mapper 持有一个按 Partition 分界的缓存区， Shuffle 数据首先写入缓存区，每当某个 Partition 的缓存满了即触发PushData。
在 PushData 触发之前 Client 会检查本地是否有 PartitionLocation 信息，该Location 规定了每个 Partition 的数据应该推送的 Worker 地址。若不存在，则向Master 发起 getOrAllocateBuffers 请求。 Master 收到后检查是否已分配，若未分配则根据当前资源情况选择互为主从的两个 Worker 并向他们发起 AllocateBuffer 指令。Worker 收到后记录 Meta 并分配内存缓存。Master 收到 Worker 的 ack 之后把主副本的
Location 信息返回给 Client。Client 开始往主副本推送数据。主副本 Worker 收到请求后，把数据缓存到本地内存，同时把该请求以 Pipeline 的方式转发给从副本。从副本收到完整数据后立即向主副本发ack，主副本收到 ack 后立即向 Client 回复 ack。为了不 block PushData 的请求， Worker 收到 PushData 请求后会先塞到一个
queue 里，由专有的线程池异步处理。根据该 Data 所属的 Partition 拷贝到事先分配的buffer 里，若 buffer 满了则触发 flush。 ESS 支持多种存储后端，包括 DFS 和 local。若后端是 DFS，则主从副本只有一方会 flush，依靠 DFS 的双副本保证容错；若后端是Local，则主从双方都会 flush。在所有的 Mapper 都结束后， Master 会触发 StageEnd 事件，向所有 Worker 发送CommitFiles 请求， Worker 收到后把属于该 Stage 的 buffer 里的数据 flush 到存储层，close 文件，并释放 buffer。Master 收到所有 ack 后记录每个 partition 对应的文件列表。若 CommitFiles 请求失败，则 Master 标记此 Stage 为 DataLost。在 Reduce 阶段， reduce task 首先向 Master 请求该 Partition 对应的文件列表，若返回码是 DataLost，则触发 Stage 重算或直接 abort 作业。若返回正常，则直接读取文件数据。
ESS 的设计要点，一是采用 PushStyle 的方式做 shuffle，避免了本地存储，从而适应了计算存储分离架构；二是按照 reduce 做了聚合，避免了小文件随机读写和小数据量网络请求；三是做了两副本，提高了系统稳定性。

容错：除了双副本和 DataLost 检测， ESS 在容错上做了很多事情保证正确性。
当 PushData 失败次数(Worker 挂了，网络繁忙， CPU 繁忙等)超过 MaxRetry 后，
Client 会给 Master 发消息请求新的 Partition Location，此后本 Client 都会使用新的
Location 地址。若 Revive 是因为 Client 端而非 Worker 的问题导致，则会产生同一个 Partition 数据分布在不同 Worker 上的情况， Master 的 Meta 组件会正确处理这种情形。若发生WorkerLost，则会导致大量 PushData 同时失败，此时会有大量同一 Partition 的Revive 请求打到 Master。为了避免给同一个 Partition 分配过多的 Location， Master保证仅有一个 Revive 请求真正得到处理，其余的请求塞到 pending queue 里，待Revive 处理结束后返回同一个 Location。
当发生 WorkerLost 时，对于该 Worker 上的副本数据， Master 向其 peer 发送CommitFile 的请求，然后清理 peer 上的 buffer。若 Commit Files 失败，则记录该Stage 为 DataLost；若成功，则后续的 PushData 通过 Revive 机制重新申请Location。
Speculation task 和 task 重算会导致数据重复。解决办法是每个 PushData 的数据
片里 encode 了所属的 mapId， attemptId 和 batchId，并且 Master 为每个 map task记录成功 commit 的 attemtpId。 read 端通过 attemptId 过滤不同的 attempt 数据，并通过 batchId 过滤同一个 attempt 的重复数据。

在 DFS 模式下， ReadPartition 失败会直接导致 Stage 重算或 abort job。在Local 模式， ReadPartition 失败会触发从 peer location 读，若主从都失败则触发Stage 重算或 abort job。

ESS 目前支持 DFS 和 Local 两种存储后端。

ESS 以不侵入 Spark 代码的方式跟 Spark 集成，用户只需把我们提供的 Shuffle Client jar 包配置到 driver 和 client 的 classpath 里，并加入以下配置即可切换到ESS方式：spark.shuffle.manager=org.apache.spark.shuffle.ess.EssShuffleManager

对 ESS 服务端进行了较为详尽的监控报警并对接了 Prometheus 和 Grafana。