做湲兔费网站视颍,公司内部网站怎么建立,适合女生做的网站主题,企业微网站与手机微信摘要#xff1a;本文整理自阿里云/数据湖 Spark 引擎负责人周克勇#xff08;一锤#xff09;在 Streaming Lakehouse Meetup 的分享。内容主要分为五个部分#xff1a; Apache Celeborn 的背景Apache Celeborn——快Apache Celeborn——稳Apache Celeborn——弹Evaluation… 摘要本文整理自阿里云/数据湖 Spark 引擎负责人周克勇一锤在 Streaming Lakehouse Meetup 的分享。内容主要分为五个部分 Apache Celeborn 的背景Apache Celeborn——快Apache Celeborn——稳Apache Celeborn——弹Evaluation 点击查看原文视频 演讲PPT 一、背景 大数据引擎的中间数据有两个来源: Shuffle 和 Spill其中最重要的是 Shuffle据统计有超过 15%的资源消耗在 Shuffle。 1.1 传统 Shuffle 的问题 下图右侧结构图是传统 Shuffle 的过程左边是 Mapper Task基于 Partition ID 对 Shuffle 数据排序然后写到本地盘同时会写一个索引文件以记录文件里属于每一个 Partition 的 offset 和 length。在 Reduce Task 启动的时候需要从每一个 Shuffle 文件里读取属于自己的数据。 从 Shuffle 文件的角度来看它接收大量并发的读请求且这些请求所读的数据是随机的这就会带来随机的磁盘 I/O。 另外一方面从下图也可以看到网络的连接数也非常多。 总结而言Shuffle 算子非常重要但是也存在一些问题 作为最重要的算子资源消耗超过 15% 高网络连接、随机磁盘 I/O 和 I/O 放大使得其不够高效 Fetch Failure 和 OOM导致其不够稳定 依赖大容量本地磁盘无法存算分离。 1.2 Apache Celeborn 的介绍 Apache Celeborn 是为了解决上述 Shuffle 的问题定位是统一中间数据服务。 Apache Celeborn 具有两个维度 第一引擎无关。官方已经实现了 Spark 和 Flink。当前我们也在做 MR 和 Tez 的集成。 第二中间数据。这里是指包括 Shuffle 和 Spill Data。当我们把中间数据全部托管它的计算节点就不需要这么大的本地盘了也就意味着计算节点可以做到真正的无状态这就可以实现在作业运行的过程中做到更好的伸缩从而获得更好的弹性和资源使用率。 Apache Celeborn 的发展史 2020 年诞生于阿里云 2021 年 12 月对外开源同年做到云上开发者共建构建多元化社区 2022 年 10 月进入 Apache 孵化器。 二、Apache Celeborn——快 Apache Celeborn 的快将从四个角度展开介绍 核心设计 列式 Shuffle 向量化引擎 多层存储 2.1 核心设计Push/聚合/Spilt 从下图可见左侧是 Apache Celeborn 最核心的设计本质是一种 Push Shuffle 和 Partition 聚合的设计。它会把同一个 Partition 数据推送给同一个 Celeborn Worker。 正常情况下每一个 Partition 的数据都会最终形成一个文件Reducer 在读取的时候只需要从 Worker 上读取一个文件就可以了。 因为 Shuffle 数据存在 Celeborn Cluster 里不需要存放在本地磁盘所以可以更好的做存算分离。另外它是 Push Shuffle不需要对全量 Shuffle 做数据排序所以也不存在写放大的问题。 第三通过 Partition 的聚合解决了网络和磁盘 I/O 低效的问题。 上图右侧的架构表明数据倾斜很常见即使在非倾斜的情况下某一个 Partition 的数据特别大也是很容易发生的。这就会给磁盘带来较大的压力。所以这边做了一个 Split 机制。简单来讲就是 Celeborn Cluster 会检查某一个文件的大小如果超过了阈值就会触发 Split也就是说这个 Partition 数据最终会生成多个 Split 文件Reduce Task 会从这些 Split 文件中读取 Partition 的数据。 2.2 核心设计异步 我们在很多环节做了异步化为的是不论在写、读还是 Control Message 过程中不 block 计算引擎本身的计算。 异步刷盘无论是双备份还是单备份在 Worker 端接收到数据后不需要等刷盘就可以发 ACK。刷盘是异步的当属于某个 Partition 的 Netty Buffer 达到某个阈值后触发刷盘从而提升刷盘效率。 异步 Commit 是指在 Stage 结束后会有一个 Commit 过程简单来讲需要让参与 Shuffle 的 Worker 把内存数据刷盘。这个过程也是异步的。 异步 Fetch 是比较常见的意思是 Partition 数据生成了文件切成很多 Chunk那么在 Fetch 的时候可以 Fetch 多个 Chunk这样就可以把 Fetch 数据和 Reduce 计算 Pipeline 起来。 2.3 列式 Shuffle Celeborn 支持列式 Shuffle写时做行专列读时做列转行。相比于行存列存具有更高的压缩率数据量可以减少 40%。 为了降低行列转换过程中的解释执行开销Celeborn 引入了 Code Generation 的技术如上图右侧所示。 2.4 对接向量化引擎 大数据计算引擎用 Native 向量化提升性能这是目前的一个共识。无论是 Spark 还是别的引擎大家都在往这个方向探索。 Gluten 是英特尔和麒麟联合发起的项目能够让 Spark 集成其他的 Native 引擎。除此之外Gluten 还做了内存管理和 Native Shuffle它的 Native Shuffle 相比原生的 Java Shuffle 更加高效但其沿用了 ESS 框架因而存在前述的限制。 当 Celeborn 社区和 Gluten 社区合作就可以将两者的优势结合这样就可以做到优化正交。 2.5 多层存储 Shuffle 有大有小对于小的 Shuffle 需要走一层网络效率是难以保障的。多层存储从通过内存缓存进行优化。 多层存储定义了内存本地盘和外部存储外部存储包括 HDFS 或 OSS设计理念是尽可能让小 Shuffle 的整个生命周期都能贮存在内存里并尽可能的落在更快的盘里。 三、Apache Celeborn——稳 有了 Celeborn 的核心设计大 Shuffle 作业在性能和稳定性上有了很大提升。Celeborn 服务自身的稳定性可以从四个角度展开 容错 快速滚动升级 Traffic Control 负载均衡 3.1 容错 如下图在容错这个层面我们做了以下工作 上图右侧描述了 Revive 机制。Client 推数据是最高频的操作也是最容易发生错误的地方。当 Push 失败我们采取了比较宽容的策略将这次推送认为是 Worker 短暂不可用只需要把将来的数据推送到别的 Worker 上就可以了这就是 Revive 机制的核心。 右侧下面的 Batch Revive 是针对 Revive 机制的一个优化。也就是说当 Worker 不可用所有往这个 Worker 上推送的数据请求都会失败那么就会产生大量的 Revive 请求为了降低这些请求的数量我们对 Revive 做了 BatchBatch 化之后就可以批量的处理错误。 关于磁盘防爆上文也提及过我们会检测单个文件大小并让其切分。另外还会检查当前磁盘的可用容量是否足够如果不足会触发 Split。 3.2 快速滚动升级 下图详细的介绍了 Celeborn Worker 需要滚动升级的时候是怎么样在不影响当前运行作业的情况下完成滚动升级的。 Worker 触发优雅停机后把状态告诉 MasterMaster 就不会继续往 Worker 上分配负载同时 Worker 上正在服务的 Partition 请求会收到一个 HardSplit 标记然后触发 ReviveClient 就不会再往这里推送数据同时会给 Worker 发一个 CommitFile触发内存的数据的刷盘。这个时候 Worker 不会收到新的负载老的负载也不会被推送内存的数据也全部都写入磁盘了。此时 Worker 把内存状态存储到本地的 LevelDB 后就可以安全退出了。重启之后从 LevelDB 中读取状态继续提供服务了。 通过以上的这种机制可以做到快速滚动升级。 3.3 Traffic Control Traffic Control 的目的是不要打爆 Worker 的内存也不希望打爆 Client 的内存。如下图所示这里面提供了三个机制 第一反压机制。从 Worker 角度来讲数据来源有两个一是 Mapper 推给它的数据二是如果开启两个副本那么主副本会往从副本发送数据。 那么当内存达到警戒线就会停止数据源头推送数据同时还需要“泄洪”把内存卸下来。 第二拥塞控制。在 Shuffle Client 端采用类似 TCP 的拥塞控制主动控制推送数据的速率避免瞬时流量把 Worker 内存打爆。 一开始处于 Slow Start 的状态推送速率较低但是速率增长很快当达到拥塞避免阶段速率增长会变慢。一旦收到 Worker 端拥塞控制的信号就会马上回到 Slow Start 状态。Worker 端会记录过去一段时间来自各个用户或是各个作业推送的数据量然后决定谁应该被拥塞控制。 第三Credit Based。用于 Flink Read 场景在 Shuffle Read 时需要保证所 Read 的数据是被 Flink 管理的。简单讲就是 Worker 把数据推给 Task Manager 之前需要拿到 Credit。 3.4 负载均衡 这里主要是指磁盘的负载均衡针对的是异构集群场景。 异构情况下机器的处理能力、磁盘容量和磁盘健康都是不一样的。每个 Worker 都会自检本地磁盘的健康状态和性能同时把结果汇报给 Master这样 Master 有一个全局的磁盘视野可以根据一定的策略在这些磁盘之间做负载分配实现更好的负载均衡。 四、Apache Celeborn——弹 使用 Apache Celeborn 的典型场景有三种完全混部、Celeborn 独立部署和存算分离。 完全混部的收益主要是提升性能和稳定性但是它的资源因为固定所以很难做到弹性。 Celeborn 独立部署Celeborn 的 I/O 和 HDFS 的 I/O 可以做隔离免去互相的影响且 Celeborn 集群具备一定弹性。 存算分离计算和存储是分开的Celeborn 集群独立部署。计算集群因为 Shuffle 变为无状态可以做很好的弹性Celeborn 集群本身也具有弹性能力存储这边也可以按照存储量收费。所以这是一个成本、性价比比较好的方案。 五、Evalution 5.1 稳定性 Spark 大作业 场景混部方式用 Spark on Yarn Celeborn。部署了 1000 台 Celeborn Worker但 Worker 资源使用量是比较少的内存大概是 30g 左右每天的 Shuffle 数据量是若干个 PB。 从上图可见这是一个非常典型的大作业有几万个并发在运行的过程中仍然非常稳定。 Flink 大作业 下图是阿里内部 Flink Batch 作业的一个截图。部署方式是 Flink on K8s Celeborn On K8s部署了 500 台 Worker每个 Worker 是 20G 的内存。 这也是一个非常大的作业可以看到它单个 Shuffle 有 680TB但运行过程也是很稳定的。 5.2 滚动重启 下图是已经测试作业运行中 Worker 的滚动重启。将一个 Worker 停掉等进程退出再重新启动。下图的时间点可见从 19 分 44 秒开始停止19 分 53 秒退出作业20 分 1 秒的时候重新启动并完成注册并继续开始提供服务。整个过程只需 27 秒作业完全没有受到影响。 另外一个用户在生产上做了滚动重启升级滚动重启 1000 台 Worker分了 10 批执行观察下来是每 2 分钟左右就可以完成一批次的重启完全不影响作业。 5.3 性能 如下图所示Celeborn0.2 和 0.3 相比 ESS 都有比较明显的性能提升同时 0.3 版本比 0.2 版本有进一步的性能提升。 下图测试的是 TPCDS对比的是 ESSCeleborn 单副本和两副本。可以看到单副本性能有 20%多的提升两副本有 15%的提升。 5.4 弹性 如下图所示用存算分离的架构部署 100 台 Worker计算是 Spark On K8s每天伸缩数万个 Pods。 加入我们 GitHub GitHub - apache/incubator-celeborn: Apache Celeborn is an elastic and high-performance service for shuffle and spilled data. 钉钉群: 41594456 微信加好友进群brick_carrier 微信公众号Apache Celeborn孵化 QA Q请问是不是把以上说的那些优化手段放在本地上会不会取得更好的效果 ASpark 有一个 LinkedIn 主导的 Magnet 的优化但因为 Shuffle 依然使用 Node Manage 管理会存在一些问题。 第一个问题是解决不了存算分离的问题第二从性能角度来讲Magnet 保留两种方式在保留了写本地 Shuffle 的同时做 Push Shuffle也就是异步读取本地 Shuffle 文件之后再去 Push 远端的 ESS而不是一边产生数据一边去 Push。 这样就带来一个问题当 Shuffle 结束的时候并不能保证所有的数据都 Push 到远端那么为了避免过长的等待时间它会强行中断这个过程。也就是最终有一部分数据被推到远端有一部分并没有。 这种情况下从控制逻辑来讲读数据的时候要先尽量去读 push 的 Shuffle如果没有的话再读本地的 Shuffle这是一个混合的过程。这个过程是有 Overhead 的。 Q请问单独部署 Shuffle 会不会带来额外的网络开销 A这其实是架构选择的问题。如果不需要存算分离不需要对计算集群做弹性扩容只是为了解决性能和稳定性的问题那么可以选择混合部署。如果需要做计算的弹性那就更倾向于单独部署。 Q请问 Spark 在执行过程中Stage 可能会失败在这种情况下怎么处理呢 A这其实是数据正确性的问题。Spark Task 重算会导致重复数据的推送Celeborn Client 也可能重复推送数据。 第一Spark 会记录哪个 attempt 成功了Celeborn 要拿到这个信息。 第二推送的每个数据都会有一个 Map IDattempt ID 和 Batch ID这个 Batch ID 就是在 attempt里全局唯一的 ID。在 Shuffle read 的时候只读成功的那个 attempt 数据第二针对这个 attempt 数据会记录下来之前读到的所有 Batch ID如果发现了之前读过的就直接忽略了。 这样可以保证既不会丢失数据也不会重复数据。 Q请问如果 Spark 或 Flink 和 Celeborn 一起去用的话如果提交 Spark 任务中间的 Shuffle 过程是 Celeborn 自己接管那个状态还是我们可以直接用 Celeborn 实现这些功能 A这是用法的问题。如果想用 Celeborn首先需要部署 Celeborn 集群第二步把 Celeborn 客户端的 Jar 拷贝到 Spark 或 Flink 的 Jars 目录第三步启动作业的时候多加一些参数。做好这几步就可以正常使用 Celeborn 了。 点击查看原文视频 演讲PPT
