★ FEATURED ARTICLE 写在开头先说一句标题里的“赫兹威客”是我手上这个完全分布式Spark测试项目的代号接这个项目前我也在本地写过不少Spark测试但真正把集群搭起来、跑通第一轮分布式任务的时候才意识到本地模式和真分布式之间差距比想象中大得多。这篇博文就把整个测试过程完整复盘一遍从集群规划、环境配置、安装部署到提交任务、查看日志、排查故障全流程给出可复现的方案。适合那些已经会写Spark代码、但还没亲手搭过真实集群的开发者参考如果你正准备在几台机器上把Spark跑成真正的Master-Worker架构这篇文章应该能帮你少踩几个坑。1. 为什么必须区分本地模式和完全分布式1.1 本地模式和完全分布式之间隔着一整条技术栈很多人在自己电脑上跑过Spark用的是local模式。local[*]会在当前进程里启动几个线程模拟Executor来执行任务跑起来很快、调试也方便。但这里有个致命的误解local模式根本没有网络通信没有真正的跨节点调度也没有真实的数据本地性。你在local模式下写得通的任务放到集群上大概率会冒出一堆新问题——连接失败、资源不足、shuffle慢、Worker掉了这些在local模式里根本不会出现。完全分布式Spark简单说就是一台Master节点负责接收任务、调度资源和监控Worker状态若干台Worker节点真正执行计算任务。比如一个Spark Streaming任务读Kafka的数据经过解析、聚合、写回存储这些计算算子会被切成一个个Task由Master调度到不同Worker上执行。这里面的关键点是Worker之间可能存在数据交互shuffleMaster和Worker之间有大量心跳和日志通信任何一个环节配置不对整个测试就得从头排查。所以做完全分布式测试的意义不是走个形式而是让任务部署、调度、资源隔离、网络通信这些生产环境要吃到的压力全部真实发生。我给自己定的验收标准很简单任务跑完日志里能看到多个Worker同时在工作Web UI里能看到多个Executor并且一个节点挂了之后任务不会整个失败——这才叫真正的分布式测试。1.2 节点规划几台机器、多少内存、怎么分工威客项目交付时客户要求至少三台节点组成集群我用的是四台虚拟机配置都不高每台2核4G跑测试足够但也能真实暴露资源不足的问题。节点角色分工如下节点名角色说明node01Master Worker承担Master调度同时参与计算node02Worker纯计算节点node03Worker纯计算节点node04Worker 测试客户端主要用于提交任务也参与计算理论上Master和Worker可以混布但正式生产环境不建议这么干因为Master要承担调度如果它所在节点被大计算任务占满心跳延迟会直接拖垮整个集群。测试集群倒是无所谓不过我也建议至少给Master节点留一个核给调度进程。内存规划上我习惯用一个简单原则任务需要的Executor内存总和加上系统自身开销不要超过物理内存的80%。四台机器每台4G我分配了SPARK_WORKER_MEMORY3g剩下1G留给操作系统和worker进程自身。这里要注意一个坑如果Worker配置的内存超过实际物理内存启动时不会报错但一旦任务申请执行器、分配堆内存系统会疯狂swap甚至直接OOM任务看起来像卡死一样。后面排查那一章会重点说。网络方面我要求四台机器的内网IP在同一网段并且/etc/hosts里把主机名和IP的映射写全保证互相能ping通。这个看似基础但很多测试失败都出在主机名解析上——Master用node01这个名字把你的Worker地址下发出去结果Worker节点自己不认识node01连接直接失败进程起得来却注册不上。所以不要嫌麻烦每一台机器都要配好完整的主机映射表。2. 环境准备把地基打牢后面才不返工2.1 JDK选型别上来就装最新版Spark是跑在JVM上的JDK版本直接决定你后面能拉起的进程。Spark 3.3.x官方支持Java 8和Java 11实测下来Java 8的兼容性最稳很多老代码和依赖也不会出幺蛾子。我当时给四台机器统一装的是OpenJDK 8不要一台装JDK8、另一台装JDK17否则等到排查问题时你会分不清到底是配置问题还是版本问题。安装完JDK之后确认JAVA_HOME环境变量。用which java拿到路径再按路径反推Java安装目录把它写进/etc/profile里的JAVA_HOME。我吃过一个亏用sudo apt install openjdk-8-jdk装完java -version正常但Spark启动脚本用的是JAVA_HOME这个变量如果它没指到正确的目录启动时直接报错说找不到Java。所以装完之后每台机器都要执行一次echo $JAVA_HOME确认输出非空这一步不能省。2.2 SSH免密登录不是省事是必要搭建集群过程中我会频繁地把配置文件和Spark安装包分发到各节点。手动一台一台输入密码当然也能做但Spark自带的start-all.sh脚本就是通过SSH去远程启动每个节点的Worker进程的——换句话说如果Master到Worker不配SSH免密登录启动集群这个动作本身就做不完整。具体操作在Masternode01上生成密钥对然后把公钥追加到所有节点包括自己的~/.ssh/authorized_keys里。习惯用ssh-copy-id命令一条命令完成ssh-keygen -t rsa -b 4096 -P -f ~/.ssh/id_rsa ssh-copy-id sparknode01 ssh-copy-id sparknode02 ssh-copy-id sparknode03 ssh-copy-id sparknode04这里强调两个细节。第一我在每台机器上都创建了专门的系统用户spark而不是用root跑集群。原因很简单用root跑Spark如果配置了HDFS或外部存储权限模型可能会被绕过去但更多时候是因为有些环境变量和路径是绑在某个用户下面的用root后面换回普通用户各种目录权限会非常混乱。第二SSH免密配置完成之后一定要逐个执行一次ssh node02 echo ok确认能够无密码登录。不要只测试Master到WorkerWorker之间的免密也配上因为有些操作要在集群节点上直接执行后续调试会省掉很多密码输入的时间。2.3 安装包版本怎么选直接选带Hadoop依赖的二进制版Spark官网每个发行版提供好几个下载包有-bin-hadoop2.7、-bin-hadoop3.2这些尾巴。虽然Spark本身可以脱离HDFS运行但如果你的测试涉及HDFS文件读写或者后续要对接外部存储选一个带Hadoop客户端库的版本更省心。我用的是spark-3.3.2-bin-hadoop3这个包解压到/opt/spark目录然后把SPARK_HOME写进环境变量同时把Spark的bin目录加到PATH。这里有一个常见的误解要澄清下载“with Hadoop”包不代表你要真的部署Hadoop集群。Spark只是打包好了HDFS客户端相关jar包方便你直接用hdfs://协议读写HDFS而不用手动引一堆依赖。在只有几台机器做Spark测试的场景下我们完全可以不部署HDFS用NFS共享目录代替分布式文件系统也能完成分布式测试。我这次就是用的NFS方案后面讲到测试时会详细说。3. 安装配置全流程手把手操作Spark3.1 修改spark-env.sh把角色分工说清楚Spark安装解压之后不能直接启动你得先告诉它谁是Master、每个Worker分多少资源、用哪套Java环境。这些全在${SPARK_HOME}/conf/spark-env.sh里配置。第一次使用先拷贝模板cp /opt/spark/conf/spark-env.sh.template /opt/spark/conf/spark-env.sh然后编辑我这次的配置如下export JAVA_HOME/usr/local/java/jdk1.8.0_202 export SPARK_MASTER_HOSTnode01 export SPARK_MASTER_PORT7077 export SPARK_WORKER_CORES2 export SPARK_WORKER_MEMORY3g export SPARK_WORKER_INSTANCES1 export SPARK_DAEMON_MEMORY1g逐项解释一下这几个变量的含义配置时最容易出问题的就是它们SPARK_MASTER_HOSTMaster进程绑定的主机名或IP。注意这里如果填localhost那么提交任务时用spark://node01:7077去访问会失败因为服务端绑定的还是localhost。正确做法是填集群内其他节点都能访问到的主机名也就是node01。SPARK_MASTER_PORTMaster监听Worker注册和Driver提交任务的端口默认7077。SPARK_WORKER_CORES单个Worker进程能拿到的CPU核数我设的是2。不要不配置让它默认拿全部核心因为Master节点上还要留一些核给系统和其他进程。SPARK_WORKER_MEMORY单个Worker进程能使用的内存总量我这里设3g。SPARK_DAEMON_MEMORYMaster和Worker自身进程daemon的堆内存一般1g足够。如果任务非常重、Web UI日志很多可以适当加大。worker实例数默认是1就是一台机器只起一个Worker进程。如果你想在一台机器上多起几个Worker来模拟更多节点可以把SPARK_WORKER_INSTANCES调大但这样会瓜分同一台机器的内存和CPU实际资源没变多反而会增加进程管理的复杂度测试阶段不建议这么干。3.2 配置workers列表决定Worker身份Spark 3.x版本里原本的slaves文件已经改名成workers别再用旧文件名。位置在${SPARK_HOME}/conf/workers内容很简单每行写一个节点的主机名node01 node02 node03 node04这里注意几点。第一写主机名而不是写IP因为Spark会根据主机名去解析保持一致才好排查。第二每行一个不要加多余的空格或者注释符号脚本解析时对这些很敏感。第三worker列表里包含node01是因为我决定让Master节点也参与计算如果你不想让Master分担计算就把node01从workers文件里移掉。不过测试阶段多一个Worker就能多一份资源放着不用挺浪费。3.3 分发配置到其它节点Master节点上的配置只是配置要让每个Worker机器真正生效必须把整套Spark环境分发过去。我用rsync做增量同步比scp效率高而且后续改配置重新分发也快rsync -av /opt/spark/ sparknode02:/opt/spark/ rsync -av /opt/spark/ sparknode03:/opt/spark/ rsync -av /opt/spark/ sparknode04:/opt/spark/分发之后有几件事要检查千万别省每台机器的/opt/spark/conf/spark-env.sh内容是否和Master一致尤其是SPARK_MASTER_HOST一定要指向node01不能各写各的。每台机器/opt/spark目录的所有者是否为spark用户。用chown -R spark:spark /opt/spark统一更改。每台机器的/etc/hosts是否有完整的集群节点映射所有节点的主机名都要在每一台上都能被解析。3.4 启动集群一条命令和三种检查所有配置就绪后在Master节点node01上执行/opt/spark/sbin/start-all.sh这个脚本会读取workers文件通过SSH逐个登录到每个节点启动对应的Worker进程同时在本机启动Master。执行完毕我习惯马上用jps检查进程状态。四台机器上应该看到这些进程node01Master、Workernode02Workernode03Workernode04Worker如果jps命令看不到可能是JAVA_HOME没配好或者Spark的bin目录不在PATH里。然后打开浏览器访问http://node01:8080这是Master的Web UI能看到当前活跃的Worker列表显示每个Worker的地址、核心数、内存和空闲资源。我之前第一次启动Web UI上只看到一个Worker后来排查发现是workers文件里node02那一行多了个空格导致主机名解析异常。所以Web UI上的Worker数量是最直观的分布式验证依据。任务真正在这个集群上跑起来后会占用Worker资源Web UI会有两套页面一套是Master管理页端口8080负责监控集群和Worker一套是每个任务自己的运行页端口4040显示这个任务内部的Stage、Executor、SQL执行情况。两者的区别要分清排查问题的时候常常两个页面来回切。4. 分布式测试从Hello World到真实负载4.1 准备一份分布式文件先解决数据读取问题本地模式里你读一个本地文件路径一切正常。到了分布式环境这个问题立刻变得尖锐如果把文件放在node01的本地磁盘然后spark-submit提交任务Driver会把任务分发给node02、node03上的Executor这些Executor在各自节点的本地路径上根本找不到这个文件。要为完全分布式测试准备输入数据至少三条路搭HDFS或者NFS共享目录或者把文件拷贝到每个Worker节点的相同路径下。我这台集群选的是NFS把node01上的一个目录/data/testdata共享出来其它三台机器挂载到同一路径。这样无论任务被调度到哪个Worker读到的都是同一份文件路径也完全一致。NFS在测试里还有个额外的好处它能模拟网络存储带来的IO开销让测试结果比local模式更接近生产环境。注意NFS并发读写会有锁和延迟所以测试时文件别太大几百MB到几个GB这个量级足够暴露问题了。4.2 第一个分布式测试跑通SparkPi并看懂Task分布先用自带的官方示例验证链路这个环节不要花太多时间优化性能目的只有一个确认任务真的被分发到多个Worker上了。执行命令/opt/spark/bin/spark-submit \ --class org.apache.spark.examples.SparkPi \ --master spark://node01:7077 \ --deploy-mode client \ --conf spark.cores.max4 \ /opt/spark/examples/jars/spark-examples_2.12-3.3.2.jar 100参数逐项说--master指定连接集群的URL格式是spark://主节点IP:7077注意不要填localhost--deploy-mode client表示Driver运行在提交任务的机器上测试阶段实在方便看日志spark.cores.max4限制这个任务最多能拿几个核心避免把整个集群资源占光最后一个数字100是SparkPi计算的分区数分区越多分布式调度特征越明显。跑完正常会打印最终圆周率近似值。但比起那个数值我更在意任务运行过程中观察到的调度情况打开任务运行日志看到INFO TaskSchedulerImpl: Adding task set这类输出后面跟着不同Executor编号说明任务确实被拆开分发给多个Worker了。如果你想再严格一点把spark.cores.max设成2然后再设成4跑两次对比耗时通常会看到资源翻倍后耗时明显下降这就是分布式调度生效的铁证。4.3 自定义测试脚本验证数据分片与shuffleSparkPi只能证明链路通畅不能证明数据分片、shuffle、数据倾斜这些问题。我写了一个简单的Python统计脚本生成大量键值数据然后做groupBy聚合和join用这种方式触发真正的shuffle。测试脚本如下from pyspark.sql import SparkSession spark SparkSession.builder \ .appName(DistributedPartitionTest) \ .getOrCreate() sc spark.sparkContext # 在NFS共享目录准备一个文本文件每行一个单词 lines sc.textFile(file:///data/testdata/words.txt) words lines.flatMap(lambda line: line.split( )) pairs words.map(lambda word: (word, 1)) counts pairs.reduceByKey(lambda a, b: a b) counts.saveAsTextFile(file:///data/testdata/output_dir) result counts.collect() print(Total keys:, len(result)) print(Sample results:, result[:5])写这段脚本时有几个点要考虑文本文件放在NFS共享目录所有节点都能访问否则某个Worker读不到数据会直接报FileNotFoundException。saveAsTextFile的输出目录不能预先存在Spark会自己创建如果同名目录已经存在任务会报错这个和Hadoop的OutputCommitter行为一致。flatMap和reduceByKey的链会让中间结果经过一次shuffle。如果你想观察shuffle的具体情况可以在Web UI的Stage详情里看到指向ShuffleMapStage的信息包括shuffle读写的字节数。跑完之后去Web UI的Executors页面看一眼应该有多个Executor条目每个都显示各自的运行时间、内存使用、shuffle字节数。如果Executors页面只有一条说明你的任务还是只在一个节点上执行这就要回头查提交参数和集群资源了。实测下来我这个20行不到的小脚本100MB输入3g内存的Worker配置分布式跑一遍大概几分钟其中大量耗时在shuffle写盘上这是分布式测试想看到的真实效果。4.4 用Web UI和日志验证分布式运行状态跑完第一个真实负载之后我习惯把测试结果记录成一张表方便跟后续调优对比测试项输入大小Worker数Executor数总耗时shuffle读写说明SparkPi无3212s0链路验证WordCount100MB3483s256MB首次shuffle键值join300MB34151s1.1GB数据倾斜严重这张表不是形式它帮助我推导出很多后续优化点。例如join测试的shuffle量异常大后来查出来是其中一个键占了一半数据明细倾斜。有了这张表我可以明确知道问题出在哪个环节再回头改并行度或者加salting处理。日志方面Spark的运行日志默认在${SPARK_HOME}/logs目录Master和Worker各自有独立的日志目录。任务提交时控制台输出的是Driver日志Executor运行日志分散在各个Worker节点上排查问题时经常需要去对应节点看日志。分布式环境里这很费时间我的做法是提前在log4j2.properties里把Spark日志级别调整一下测试阶段把org.apache.spark的日志级别设为INFO关键调度信息就能看到但别开DEBUG那个级别日志量非常大反而会拖慢任务。5. 常见问题与排查合集5.1 命令找不到JAVA_HOME和PATH的连锁反应启动集群时报“找不到java”或jps命令没有任何输出大概率是JAVA_HOME没配到/etc/profile或当前shell没执行source。我在node03上就犯过这个错所有节点的spark-env.sh都指定了JAVA_HOME但是node03的/etc/profile里没写这个变量SSH远程启动Worker时环境变量不完整脚本从$PATH找不到java启动失败。排查方法也简单:手动ssh到报错节点执行echo $JAVA_HOME如果没有输出就补配并source /etc/profile。5.2 Worker注册不上Master主机名解析和防火墙启动集群后Master的Web UI上Worker列表不完全缺了某台节点或者Worker进程反复处于DEAD状态。典型原因是Worker节点无法解析Master的主机名。我通过ssh node02 ping node01来验证如果ping不通检查两台机器的/etc/hosts。Spark默认端口被防火墙拦截。7077是Master的调度端口8080和8081是Web UI端口4040是任务监控端口。测试环境最直接的办法是临时开放这些端口或者直接关掉防火墙。生产环境要新建安全组规则别用关防火墙这种粗暴方式。我排查时还有一个习惯直接看Worker日志。路径是/opt/spark/logs/spark-spark-org.apache.spark.deploy.worker.Worker-1-node02.out日志里会明确写“Failed to connect to master”还是“Master has changed”。如果没有看到任何日志先怀疑SSH没把Worker进程拉起来。5.3 任务提交成功但一直Pending资源分配和core配额spark-submit命令不报错但任务状态一直停在Waiting或Pending打开Web UI的详情页会发现任务在等Executor资源。这种情况通常不是代码问题而是资源不够分。常见细节集群所有Worker的内存都被其他任务占满任务申请不到额外内存。spark.cores.max设置值大于当前空闲核心数调度器无法凑齐这么多核。Worker进程实际可用内存小于任务申请的内存导致一直拿不到Executor。我用过一个最小的资源计算公式一个Executor能运行时需要的内存至少是spark.executor.memoryspark.executor.memoryOverhead后者默认是10%或者384MB两者取最大值。如果spark.executor.memory是2g那实际每个Executor要占差不多2.2g。假设Worker只有3g内存一个Worker同时只能跑1个Executor。排列组合一下就知道为什么你明明有两个Worker却只拿到一个Executor了。5.4 任务跑一半Executor失效内存溢出和心跳超时测试过程中非常容易遇到Executor在某一个Stage直接挂掉任务会反复尝试重新调度这个Stage最后日志里出现ExecutorLostFailure。这种情况大概率是Executor JVM堆内存溢出。处理方向调大spark.executor.memory但同时注意不要超过Worker的3g配额。减少单个任务的分区并发度比如把spark.sql.shuffle.partitions从默认200调小减少同时打开的Shuffle文件数。检查是否存在数据倾斜某一个分区比其它分区大太多处理不过来。这时候加资源不如改算法比如给key加随机盐再聚合两轮。另外一个杀招是心跳超时Master偶尔会因为GC导致心跳延迟误判Worker失联。这种情况我会调大超时时间为120秒--conf spark.network.timeout120s --conf spark.executor.heartbeatInterval30s注意不能只调一边心跳间隔和超时时间要匹配心跳间隔太短反而增加网络压力。5.5 输出目录已存在可重复测试的一个硬伤第一次跑saveAsTextFile成功第二次再跑同样代码报“Output directory already exists”。这是Hadoop生态里的老问题Spark为了保证数据一致性不允许覆盖已有输出目录。测试阶段我一般用temp清理或每次在输出路径末尾加上时间戳import time output_path file:///data/testdata/output_{}.format(int(time.time())) counts.saveAsTextFile(output_path)这种方式在自动化测试里很实用不用每次手动清理目录。6. 写在最后一点个人经验搭建完全分布式Spark测试集群这件事真正的门槛不在装包和改配置而在你能不能理解每台机器扮演的角色能不能在出现问题时顺着日志和Web UI找到根因。我踩过的坑具体来说有两个印象最深。第一个是“资源算账”很多人装完集群一跑发现任务全挤在一个节点上或者干脆一直Pending原因就是Worker内存设置和Executor内存参数没搭好。第二个是“主机名解析——我因为一篇hosts文件没更新浪费了一上午时间最后发现只是拼写问题。如果你也想自己搭一套测试集群我的建议是先从三台最基础的机器开始用最朴素的配置把链路跑通再一步步加节点、加数据量、加并发。别一上来就追求完美集群那样只会让问题堆在一起很难定位。本文用的所有例子和配置都是我自己在实际测试中用过的参数未必适合你的硬件但排查思路和方法是通用的。希望这篇把你带过一遍流程的文章能让你少走几步弯路真正把分布式Spark跑起来。
阅读完成 · 觉得有帮助？
