简介《云计算之分布式计算》PPT是一份面向云计算初学者、高校学生及技术培训讲师的系统教学课件围绕分布式计算在云计算中的核心地位解释其如何通过网络协同多节点解决海量数据处理难题。内容先由大数据时代背景切入引用加州大学《多少信息》、IDC《Extracting Value from Chaos》及《纽约时报》数据说明数据爆炸带来的挑战再梳理分布式计算的两大分类——批量计算与实时计算并对照Google与Apache的典型实现包括GFS/HDFS文件系统、BigTable/HBase数据库、MapReduce/Hama计算框架、Percolator增量更新、Dremel与Tenzing/Hive查询分析等帮助读者建立完整技术地图。资源仅含1个pptx文件压缩包约840KB小巧聚焦便于下载后直接用于教学或自学。PPT以MapReduce天气统计工作流为例演示了从Master任务分解到Slave并行处理的完整过程页面层次分明、图表直观既适合课堂讲解也适合作为分布式计算入门分享的参考素材。目前已有106人学习浏览。1. 云计算里的分布式计算不是加机器就能变快云计算里的分布式计算不是把任务丢到多台机器上就能变快那么简单。我见过太多团队在这个点上翻车数据量刚到几百GB就拉了一个8节点Spark集群结果是任务调度花了半小时、计算只用了十分钟整体比单机还慢了好几倍。标题里的“云计算之分布式计算”这类PPT在团队里通常用来给新人做入门分享但如果只停留在概念层新人上手后依然会踩同样的坑。这篇笔记想把它拆成能直接用的东西什么样的任务适合分布式、该选哪套框架、参数怎么调以及出了问题去哪里排查。适合正在做数据管道、批处理或AI训练任务的工程师尤其是那些被“分布式很高级”误导、想找一条可靠落地路径的人。2. 先把底层的两件事分清资源池与协作关系2.1 分布式计算不是多核计算的放大版很多人对分布式计算的第一印象是“多核计算的放大版”这是最容易埋雷的误解。多核共享一块内存进程之间通信走总线快且便宜而分布式计算里节点之间只能通过网络交换数据带宽和延迟差了至少一个数量级。一旦任务切分没考虑数据本地性性能瓶颈会迅速从“计算”转移为“搬运”。我早年做过一个稠密矩阵的并行化改造把矩阵切成小块分配给多个节点每个节点算完自己那块后再相互交换边界数据做合并。结果很丢人4个节点跑出了比单机多核还慢的成绩。问题出在矩阵切块后在迭代中需要频繁交换边界行网络传输量远大于计算量。后来改成按行分块每个节点只处理本地连续行只在迭代结束后做一次规约性能才真正提上来。这件事给了我一个判断标准当一个任务的子步骤之间需要反复交换中间数据它就不适合做分布式拆分至少不能按数据块拆。真正适合分布式的是“数据可分区、结果可合并”的计算比如日志清洗、统计聚合、批量推理。判断能不能分布式先看数据能不能切再看切了之后需不需要来回传。2.2 三个特征决定任务能不能拆可拆分、可调度、可容错分布式计算能不能落地核心看三个特征是否满足。可拆分。一个任务能被切成足够细的子任务而且这些子任务之间没有强依赖或者依赖链很短。如果子步骤之间是严格串行的比如UDF里每一步都要用上一步的全量输出那么拆分的意义就非常有限。拆得越细调度开销越大所以还要看拆分粒度是否大于调度成本。可调度。子任务数量要相对于节点数量足够多否则调度器无活可干。8个task分给16个节点一半节点空转集群利用率惨不忍睹。理想情况是子任务数量是节点数的几倍到几十倍让调度器有弹性空间去做负载均衡。这就像排流水线工位数必须多于工人数才能让工人一直有活干。可容错。分布式环境下节点是廉价硬件随时可能挂。框架需要能检测到节点失败并重新调度任务但重试会带来另一个问题重复计算。如果任务不是幂等的比如写数据库时没有去重标识重试一次就多写一份数据。所以设计分布式任务时任务本身的输出最好带一个幂等键把“至少一次”的语义安全地降级成“恰好一次”。2.3 云资源池与分布式框架一个管资源一个管任务这里有必要把云计算和分布式计算的关系说透。云计算提供的是资源池云主机、容器、存储桶、配额、弹性伸缩解决的是“机器从哪里来”。分布式计算解决的是“任务怎么跑”它跑在资源池之上负责把任务切分、调度、容错。云计算运维工程师日常盯着资源池的CPU、内存、网络以及配额是否够用能保证的是节点的健康和可用性但没法回答“为什么任务变慢了”。分布式计算框架负责的是任务在资源池上的切分和编排。所以一线排查时常见分工是节点日志报错先找云计算运维确认资源状态任务级失败或倾斜再回到分布式框架层分析。把这两层混在一起容易排查方向跑偏浪费很长时间。3. MapReduce、Spark、Ray三种主流程架怎么选3.1 离线批处理MapReduce与Spark的取舍要看内存预算MapReduce是第一代分布式计算框架它的设计哲学是简单可靠Map阶段把输入拆成小块并行处理中间结果落到本地磁盘Shuffle排序后交给Reduce阶段。落盘让每一步都可以独立恢复节点挂了只重算对应分片容错非常好。它的代价也很明显每一步都落盘IO开销巨大对延迟敏感的作业不适合。在纯MapReduce里我最常调的参数是mapreduce.reduce.shuffle.parallelcopies它控制Reduce阶段从多个Map节点并行拉取数据的并发数。默认值通常只有5在机器多、网络带宽充裕的机房里把它调到核心数的一半Shuffle耗时能明显下降。但这不是万能调节旋钮拉取并发太高会把网络打满。Spark比MapReduce快的原因在于把中间结果留在内存配合DAG调度把多个计算步骤流水线化减少落盘。但Spark的内存优势有个前提executor内存要够。如果一个executor只分到1GB数据一超过阈值就会溢写到磁盘速度优势直接消失甚至比MapReduce还慢。我经手过几个从MapReduce迁移到Spark的ETL任务速度提升通常在2到5倍之间但前提是spark.executor.memory和spark.sql.shuffle.partitions都调过后者决定Shuffle后生成多少分区数据量大而分区数太小每个分区要处理的数据太多内存吃紧。3.2 动态任务图与AI训练选Ray而不是硬上SparkRay和Spark的最大区别在于任务模型。Spark面向的是有序的数据处理流程Stage之间是静态的依赖关系Ray面向的是动态任务图任务之间可以通过对象引用直接传递数据任务依赖在运行时自然形成不需要提前规划成Stage。在AI训练和超参搜索场景里这种差异非常明显。我用Ray做过一组并行超参搜索10组超参组合每组训练一个小模型每轮训练结束要根据当前结果动态调整下一轮的参数组合。在Spark里表达这种“算完再决定下一步跑什么”的控制流很别扭因为Spark的算子层面不擅长表达细粒度的循环和动态分支。而Ray的ray.remote函数天然支持这种调用每个训练任务是一个独立的远程函数用.remote()提交后返回一个ObjectRef需要时ray.get()取结果。Ray的另一个优势是Actor模型适合有状态的在线服务场景。比如把训练好的模型封装成一个Actor常驻内存外部请求直接调用Actor的方法避免了每次推理都冷启动一个模型的尴尬。这在实际生产中省掉了很大一部分重复加载开销。3.3 一次讲清选型边界三套框架的对比与配置要点很多人在项目选型时按名气来这是不靠谱的。我一般会画一张表把框架的适用场景和不适用场景列清楚再结合团队已有的运维习惯做决定。框架最擅长最不擅长核心配置项MapReduce超大批量、对延迟不敏感的离线任务需要严格落盘容错延迟敏感的交互式查询、需要反复迭代的计算mapreduce.reduce.shuffle.parallelcopiesSpark内存充足的ETL与SQL分析需要流式批次处理任务依赖频繁变化、需要动态分支的流程spark.executor.memory、spark.sql.shuffle.partitionsRay动态任务图、并行超参搜索、强化学习、有状态服务纯SQL批处理、离线报表num_cpus、object_store_memory表里的每一项都可以展开成决策问题任务依赖是静态的吗选Spark依赖是运行期才确定的动态结构选Ray对容错的要求是“必须落盘能恢复”选MapReduce。选型之后还要看团队里有没有人扛得起运维。Spark生态大、文档多招人相对容易Ray上手快但生产化还需要自己补监控和部署能力。别只听框架社区的声音要结合自己团队的执行能力。4. 本地跑通最小分布式计算从multiprocessing到Ray4.1 先用multiprocessing跑通多进程骨架理解并发与开销在把任务真正搬上Ray或Spark之前我习惯先在本地用multiprocessing跑一个最小骨架。这能直观感受并行度的作用也能暴露任务太小时调度开销反而盖过收益的问题。import math import multiprocessing as mp import time def worker_func(start, end): 对一个区间内数字做平方根累加模拟 CPU 密集计算。 total 0.0 for i in range(start, end): total math.sqrt(i) return total def run(workers: int, n: int 2_000_000): 把 [0, n) 切分成 workers 段每段交给一个进程。 chunk n // workers tasks [] for i in range(workers): start i * chunk end n if i workers - 1 else (i 1) * chunk tasks.append((start, end)) start_time time.time() with mp.Pool(processesworkers) as pool: results pool.starmap(worker_func, tasks) return time.time() - start_time, sum(results) if __name__ __main__: for w in (1, 2, 4, 8): elapsed, total run(workersw) print(fworkers{w}, elapsed{elapsed:.2f}s, total{total:.2f})这段代码的核心是Pool.starmap它把参数元组列表按顺序分发给进程池里的worker。chunk n // workers把原始数据切成workers段最后一段用end n兜底避免整除误差漏掉数据。进程数怎么设是第一个坑。很多人直接取multiprocessing.cpu_count()把所有逻辑核都用满结果CPU温度飙升监控进程还抢不到时间片。我一般按物理核数减一设留一个核给操作系统和监控线程。另一个需要观察的是当n很小、比如只有20万时workers8往往比workers1还慢因为进程创建和任务分发的固定开销超过了并行收益。这个现象在后面的分布式集群里会被放大成灾难先在本地体会一次很值。4.2 换到Ray用任务依赖替代文件传递multiprocessing解决的是单机多进程并行任务之间通过参数传递数据结果也要全部回收。一旦节点数超过一台机器这套模型就走不通了。Ray的模型更接近真实的分布式调度每个任务是一个远程函数提交后立刻返回一个ObjectRef任务间的数据传递通过ObjectRef引用完成不需要写中间文件。import math import time import ray ray.remote def worker_func(start: int, end: int) - float: 在 Ray worker 上执行的 CPU 密集任务。 total 0.0 for i in range(start, end): total math.sqrt(i) return total def run(workers: int, n: int 2_000_000): chunk n // workers futures [] for i in range(workers): start i * chunk end n if i workers - 1 else (i 1) * chunk futures.append(worker_func.remote(start, end)) start_time time.time() results ray.get(futures) return time.time() - start_time, sum(results) if __name__ __main__: ray.init() # 本地启动一个调度器连接已有集群改成 ray.init(addressauto) for w in (1, 2, 4, 8): elapsed, total run(workersw) print(fworkers{w}, elapsed{elapsed:.2f}s, total{total:.2f})worker_func.remote(...)调用不会阻塞它会立即返回一个ObjectRefray.get(futures)才真正等待所有任务完成并取回结果。这样写的好处是任务之间若存在依赖比如任务B需要任务A的输出直接把A的ObjectRef传给B即可Ray会在后台等待A完成后再调度B。相比把中间结果写到共享存储再读取这种引用传递大幅降低了IO开销。ray.init()不传参数时Ray会以本机所有可用CPU作为调度资源。如果本机同时跑着Web服务建议显式限制ray.init(num_cpus4)避免任务把在线服务的CPU抢光。连集群时用ray.init(addressauto)Ray会读取节点配置文件自动接入已有集群。需要注意ray.init(num_cpus...)限制的是调度器可见的CPU数量不是物理限制调度超了依然会排队等待。4.3 两个必调参数并行度与对象存储内存跑通骨架之后正式任务要盯两个参数。并行度。Ray里对应num_cpusSpark里对应spark.executor.instances乘spark.executor.coresMapReduce里对应map和reduce的task数。经验公式是CPU密集任务写物理核数IO密集任务写物理核数的2到4倍。并行度设低了节点空转设高了线程切换开销上升内存也可能被打满。如果任务内部还有嵌套并行比如每个worker又调用一次multiprocessing.Pool里外两层并行度相乘是爆炸的就必须去掉一层。对象存储内存。Ray的object_store_memory默认大约是物理内存的30%这个值对多数任务是够的但遇到每个任务都产出大对象的场景就麻烦。当对象存储满了Ray会把老对象驱逐如果后续任务还在引用它就会触发重新计算表现为任务运行时间突然拉长日志里出现大量“Object Lost”或重试记录。我一般先把object_store_memory设到物理内存的50%跑一轮看Dashboard里的对象淘汰次数再往下降。这个参数不是越大越好给操作系统留少了系统OOM会把Ray整个拖垮。提示不管是Spark还是Ray加并行度前先跑一次单机基线。单机都跑不动分布式只会把瓶颈放大。5. 分布式计算踩坑记录3个典型问题与排查方法5.1 小任务被调度成大任务分布式反而变慢现象一个单机脚本跑5分钟的数据清洗任务搬到8节点Spark集群后跑了45分钟。任务本身没报错但整体耗时反而长了一个数量级。原因分布式框架有固定开销。任务启动要注册、调度器要分配资源、输入数据要分片并序列化、每个task要拉取依赖这些开销在小任务里占了绝对大头。任务总计算量只有几分钟时调度开销远大于并行收益。解决先做单机压测再决定要不要上集群。判断依据是单机耗时长不长。我一般按两个阈值控制数据量小于1GB或单机耗时低于5分钟优先用单机多进程不开集群数据量在1GB到100GB之间单机算不动但任务依赖是静态的上Spark任务依赖要动态跟进或需要频繁重试用Ray。另一个常用做法是在进入分布式框架前先做一轮数据裁剪把小表、小任务提前过滤掉不是全量塞进Spark里做filter。还要检查一个容易被忽略的点集群的spark.default.parallelism或Ray的num_cpus是不是设成了节点数的倍数而非数据量的函数。并行度要跟着数据量走不是跟着机器数走。10GB数据配1000个partition通常很合理但如果你只有8个executor1000个partition意味着每个executor要串行处理一百多批启动和切换开销同样不低。5.2 数据倾斜节点之间负载落差太大现象一个统计任务在10节点集群上跑从监控看始终只有一两台机器的CPU接近100%其余机器的CPU占用不到20%。整体耗时拖到正常预估的三倍以上。原因数据按某个字段做Hash分片时热点字段把大量数据分到了同一个分区。最常见的场景是按用户ID分桶有些超级用户产生的数据量是普通用户的几百倍那个桶所在的worker就累死其他worker闲死。解决换分片键是最直接的选一个分布均匀的字段做key。如果数据本身只有一个热点字段比如都是按用户ID做聚合那就做两阶段聚合。第一阶段给key加一个随机盐把热点key打散成多个临时key并局部聚合第二阶段把盐去掉再做一次最终聚合。代码思路如下import random from collections import defaultdict def two_phase_aggregate(items): partial defaultdict(float) salt_count 8 for key, value in items: salted f{key}#{random.randint(0, salt_count - 1)} partial[salted] value result defaultdict(float) for salted_key, value in partial.items(): key salted_key.split(#)[0] result[key] value return result盐的粒度random.randint(0, 7)决定了打散强度。热点越集中盐的个数越多但也不能无限多否则第二阶段合并的耗时也上去了。一般建议盐的个数和并行度接近这样每个worker都能分到热点数据的一小片。排查时专看两点一是Spark UI里各个task的Duration分布如果有单个task耗时是均值的十倍以上基本就是倾斜二是看executor的GC时间和Shuffle读写量某个executor的Shuffle Read量远大于其他executor数据分配不均实锤。加盐后要复查这两个指标确认打散生效。5.3 共享存储的写放大中间结果不该都往云盘写现象所有worker运行时同时往同一个云盘目录写中间结果任务整体越来越慢偶尔还报“文件已存在”或“目录非空无法创建”。原因共享存储的元数据服务器扛不住大量并发的文件创建请求。每个worker每处理一批数据就创建一个新文件上百个worker同时写每个文件创建都触发一次元数据锁和目录同步IO放大效应非常严重。这类问题在做云计算运维的朋友那里尤其常见日志里全是连接超时以为是网络问题查到最后是共享存储被打瘫了。解决把“中间结果写共享盘”改成“中间结果写本地最终结果写共享”。每台节点都有本地数据盘中间产物落在/tmp或/data/local_tmp任务结束前再合并写回共享存储。Spark里可以直接配置spark.local.dir/data/local_tmp如果一定要向共享盘写那就减少文件数。Spark落地前用coalesce(n)把分区数降下来再写而不是让几千个小文件直接落到共享盘上。文件从几千个降到几十个元数据压力能减少一个数量级。这个操作在数据链路末尾做一次就够了不要在中间每个Stage都合并否则会牺牲并行度。6. 验证扩展比的两个脚本别再说“加速不明显”这种玄学话“分布式加速不明显”这句话很多团队没跑过一个定量脚本就开始凭感觉下结论。我自己固定下来的验证方式是用固定负载测试不同worker数量下的耗时再算扩展比。import time def bench(workers: int, n: int 4_000_000): # 复用第 4.2 节的 run 函数不同 worker 数跑同一份数据 elapsed, _ run(workersworkers, nn) return elapsed base_time bench(workers1) print(fworkers1, time{base_time:.2f}s, speedup1.00) for w in (2, 4, 8, 16): t bench(workersw) speedup base_time / t print(fworkers{w}, time{t:.2f}s, speedup{speedup:.2f})判定标准我用两条从1个worker到2个worker加速比如果能到1.5以上说明调度链路是健康的到4个以上时加速比开始低于理想值的60%说明任务里存在串行部分或网络等待。如果从1到2加速比连1.3都不到说明任务本身的开销就集中在单点继续加节点没有意义。第二个脚本看资源占用率。任务跑着的时候观察CPU整体利用率如果CPU利用率不到60%瓶颈在等待IO或网络不在地核数量如果CPU快满了但加速比上不去可能是锁竞争或串行段卡着Amdahl定律。这两个判断脚本加起来不到20分钟就能跑完能省下后面几周的调优加班。我现在的习惯是任何分布式方案上线前先跑这两个脚本扩展比不合格就退回单机多进程合格了再考虑上集群。这不是什么高深技巧只是把“觉得快”“觉得慢”换成了一组能复现的数字。希望帮到你。本文还有配套的精品资源点击获取