阿里云EMR 2.0：兼容开源，贡献开源，超越开源

摘要：本文整理自阿里云资深技术专家吴威(无谓)在 阿里云EMR2.0线上发布会 的分享。本篇内容主要分为三个部分：
1.兼容开源阶段
2.贡献开源阶段
3.超越开源阶段

兼容开源阶段

开源这个词在最近这几年异常的火爆，各行各业的各个厂商纷纷宣布拥抱开源并且支持开源生态。尤其在大数据这个领域，开源技术已经成为了推动整个大数据技术演进和行业发展的最重要的一股力量，同时开源技术栈也成为大数据行业的一个技术标准。阿里云EMR 作为开源大数据平台，集成了众多主流开源引擎比如 Spark、Flink、 StarRocks 等，这些引擎共同基于 EMR 计算资源底座以及数据湖存储底座，在适配阿里云生态技术栈的同时，兼容开源是 EMR 团队的一项重要工作。

事实上，阿里巴巴集团在十三年前就开始投资开源大数据领域，经过十几年的发展和进步，现在我们的开源大数据平台已经成为阿里巴巴大数据技术体系中的中坚力量。

接下去我简单分享一下阿里开源大数据技术的演进路线。

2008和2009年，阿里巴巴最主力的业务线是淘宝和天猫，其上的电商业务爆发式增长，同时业务数据也出现了大爆发，我们需要大数据技术去处理海量的业务数据，当时一度出现技术跟不上业务发展的节奏。我们在2009年选择了 Apache Hadoop 技术去支撑大数据分析业务，上线的第一个集群就达到200台规模，并且在一年内快速增长到1000台，在2014年具备跨数据中心的集群管理能力，单个开源 Hadoop 集群达到过万台的规模。开源大数据技术对阿里巴巴核心业务的发展起到了非常关键的支撑作用。这个阶段阿里巴巴是开源的受益者。

贡献开源阶段

2014年之后，因为我们在内部业务上积累了大量开源使用经验，做了不少最佳实践沉淀，我们的开源团队也转到了云上，并于2016年在阿里云推出了 EMR 产品，发现云上有更加旺盛的开源大数据的需求。

与此同时，2015年在阿里巴巴搜索推荐广告业务线上，数据实时化的需求非常强烈，我们希望搜索引擎能够搜索到实时更新的宝贝并根据用户的实时行为进行推荐，当时我们选择了 Apache Flink 作为新一代的实时计算引擎，在2016年将其上线并得到了非常好的效果。在2018年的时候，Flink 和 EMR 一样开始上云。2019年我们不仅收购了 Flink 在欧洲的创始公司，还把阿里巴巴积累多年的 Flink 分支 Blink 完全贡献给了 Apache 社区。同时我们也不断地对 Flink 进行了技术创新，推出了 Flink CDC、Flink Table Store 等新的开源技术项目。

回过头再看 EMR，自2016年在公有云上线之后，已经服务了数千家中小企业，支持他们在云上更好地使用开源大数据。目前 EMR 也经过了技术升级， 从经典的 Hadoop 架构升级到了数据湖存算分离的架构。与此同时我们也保持了整个开源大数据平台的开放性，跟国内外知名的开源大数据厂商比如 Elasticsearch、Cloudera、Databricks 等建立了密切的合作伙伴关系，并且联合推出了开源大数据的产品，在云上共建开源大数据生态。

以 Apache Celeborn 项目为例，可以看到阿里巴巴从2009年开始使用开源大数据技术，经历了十数年回馈和共建，最终希望进入到一个引领开源，推动开源发展的阶段。

2022年10月份阿里巴巴向 Apache 孵化器捐赠了 Celeborn 项目（也就是原来的 EMR Remote Shuffle Service 项目 ），这是在阿里云上诞生的第一个 Apache 孵化项目。随着大数据上云趋势越来越明显，云原生架构和理念也在不断强化和推行，比如存算分离架构等都是云上特有的架构属性。在此技术背景之下，我们发现在 Spark、Hive 、Flink 等都有数据 Shuffle 的需求，并且因为云原生架构上没有 Hadoop YARN NodeManager 等服务，无法很好的支持 Shuffle 场景，也无法实现动态资源伸缩等核心功能。因此，阿里云EMR 率先提供了 Remote Shuffle Service，用一套数据 Shuffle 服务支持所有大数据计算引擎。Remote Shuffle Service 项目诞生后，又吸引了以小米为代表的多家阿里云上公司的兴趣。在2021年12月，我们和这些阿里云客户一起将这个项目开源，之后吸引了更多如 Shopee、网易等企业的开发者加入，这就是云带来的变化，云与开源结合后产生了化学反应。为了让更多公司参与共建，让项目产生更大的影响力，我们决定将这个项目捐献给 Apache 基金会，并且正式命名为 Celeborn，从孵化器项目起步，也希望 Celeborn 能够成为 Apache 的顶级项目。

超越开源阶段

在兼容开源和贡献开源之外，EMR团队也在基于开源技术打造超越开源的企业级计算引擎，下面我以EMR Spark为例介绍他的企业级功能。首先是性能，EMR Spark 多次打破大数据领域的行业记录。

比如在 Sortbenchmark 组织的 CloudSort 竞赛中，需要完成 100TB 数据排序的同时，花费的成本要尽可能的少，EMR Spark 在阿里云上实现了每 TB 仅花费 1.44美元的成绩，比之前在海外云平台上的最好成绩提升了3倍以上。 在另外一个大数据知名 Benchmark TPC-DS 中，EMR 同样多次刷新最好成绩，并且是通过 TPC 官方 Benchmark 认证的第一个公共云产品，EMR Spark 结合 JindoFS 实现的存算分离架构在性能和性价比上都有极大的提升，并且在功能上覆盖了 TPC-DS benchmark 数据加载、数据更新、串行查询、并行查询、数据可靠性测试等全流程，SparkSQL 的数据正确性通过 TPC 组织的审计。在数据规模上，EMR Spark 是首个通过官方认证的支持最大规格 100TB 数据集的系统。

接下去展开看一下 EMR Spark 的性能优化点。我们对计算引擎涉及到的几乎所有方面都有改进。比如在 SQL Runtime 方面，EMR Spark 从2019年就开始调研 Native codegen 以及向量化技术，通过将 Spark SQL 的执行流程或算子换成 Native engine，相对于 Spark 原生的 Java Codegen 引擎性能提升非常明显，Native engine 可以支持 TPC-DS 所有 SQL，有着非常高的 Spark SQL 算子覆盖度。 Spark Native engine 将在明年初通过内测版的形式开放给更多的客户。另外在 JSON 解析上，EMR Spark 通过 SIMDJSON 技术可以将复杂 Json 解析速度提升5倍以上，可以极大的加速 JSON 日志类数据的清理和 ETL 流程。

在 SQL 优化器上，我们实现了全新的 Join Reorder 算法 ，在超过10张表 Join 的场景 SQL Planner 效率有数倍提升，我们在动态分区裁剪的基础上实现了基于 BloomFilter 的过滤特性，进一步减少了大表的数据扫描量；另外，优化器还支持 Hive 表的分区统计信息和主外键约束信息，可以提供更加精细化的优化规则。

EMR Spark 针对数据湖存储在读写上都有深度优化，配合 JindoSDK 解决对象存储在大数据场景下的痛点。

在数据 Shuffle 的场景，配合前面提到的阿里巴巴开源项目 Celeborn，EMR Spark 支持 Push Based Shuffle，将 Shuffle 数据推送到远端的 Remote Shuffle Service 上，通过服务端的数据聚合可以极大的减少磁盘的 IO 数量，这项能力对大数据常用的 SATA 盘构建的集群特别友好，对于 Shuffle 数据量超过百TB的巨型作业，Shuffle read 时间可以降低数倍。同时 EMR Spark 还支持列式 Shuffle 能力，配合数据压缩功能可以将 Shuffle 数据量减少一倍以上，这样可以进一步降低 Shuffle 的网络 IO 开销，提升集群的整体性能和吞吐量。

在性能提升之外，EMR Spark 提供了企业级的诊断和调优能力。EMR 2.0 推出了 EMR Doctor 产品，它的功能就包括 Spark 作业的智能诊断，Doctor 使用 Java 探针技术，可以对 Spark 作业做异步数据采集，并将采集到的 Spark job、stage、task、sql 等信息统一汇总到 EMR 元数据仓库进行离线数据分析。Doctor 提供了集群的计算健康分，对每个作业健康状态做打分，并提供优化建议。基于 EMR 的这套机制，我们还为 Spark 提供了基于历史信息的优化能力（也就是 HBO）。针对 Spark 作业常用的内存、partition 数和 AQE Shuffle 参数，HBO 首先会采集每天作业执行过程中的各项指标，并在元数据仓库里进行分析，在应用特定的优化规则后产出更优的作业参数。假设每天的作业数据量变化不大，那么同样的 SQL 通过历史信息调优后可以做精确的 Planner。在典型的1TB TPC-DS Workload 中，使用 HBO 优化规则后可以将 SQL 查询性能提升 28%。

针对前面提到过很多次的云原生架构，在阿里云上的K8S服务主要由 ACK 和 ECI 两个产品提供，EMR Spark 支持 on ACK 形态，并且实际解决了非常多的架构适配问题，落地了不少生产型客户。EMR 采用虚拟集群的方式，直接在用户已有的 ACK 集群安装 Spark 相关组件。同时也提供独立的 RSS（Apache Celeborn）集群类型，可以优化容器环境下的 Shuffle 场景。Spark on ACK 支持常驻节点，可以运行 Spark operator 或者 Spark History Server 等常驻服务，也支持 ECI 实例部署方式，将 Spark 的 Executor 调度到 ECI 弹性实例上，可以进一步降低资源使用成本。

在 ACK 场景下 Spark 和 Celeborn 的集成可以让 Spark 具备传统 Hadoop YARN 集群里常用的动态资源伸缩能力，提升资源利用效率。异步非阻塞和高性能的列式 Shuffle 可以提升作业性能。使用高性能的本地化多副本存储以及多层存储方案可以提升 Spark 作业的稳定性。所以 EMR Spark 的云原生解决方案在功能、性能、稳定性以及成本和效率上都有考虑，是一套成熟的方案。

EMR Spark 是数据湖架构下的核心计算引擎，和多个模块有着深度整合。

Spark 支持阿里云全托管的 HDFS 服务 OSS-HDFS，并在数据读写接口上有非常多的优化，包括前面提到的 parquet 和 orc 等格式的 native reader，以及在数据提交阶段的 job comitter 优化。另外，对于大数据中经常遇到的小文件问题，EMR Spark 提供了小文件合并能力，可以降低对存储系统的压力，同时提高读表的性能。

Emr Spark 还支持了数据湖构建也就是 DLF 产品，集成了 DLF 统一数据湖元数据和权限控制功能，借助 DLF 可以对 Spark 表进行存储分析和生命周期管理，对湖表格式进行自动化管理，比如小文件合并、排序、索引等功能。

在数据湖格式的集成上，EMR Spark 扩展了对 Delta lake 和 Hudi 的支持。在 Delta Lake 上我们实现了基于固定力度的缓慢变化维功能，借助 Delta Lake 的 checkpoint 和 Time travel 的能力，通过一些特殊的标记保留维度表的历史快照，并且不需要增加额外的冗余数据，通过 Spark 查询这些维度表，同时可以提升查询效率。

针对 Hudi，我们实现了一个全托管的 Hudi Metastore 服务，可以将分散的 Hudi 元数据进行集中管理，在解决了元数据一致性问题之后，EMR Spark 可以基于 Hudi Metastore 进行查询加速；除此之外，EMR 团队也是 Hudi 社区 CDC 功能的贡献者，在 EMR Spark 版本里已经提前支持了 Hudi CDC 能力。

以上就是 EMR Spark 在开源软件之上提供的企业级能力。阿里云EMR 团队将继续沿着兼容开源、贡献开源以及超越开源道路前进，为云上客户提供最佳的开源大数据解决方案。

原文链接

本文为阿里云原创内容，未经允许不得转载。