Standalone集群实战：PySpark作业提交与spark-submit调度

这篇日志写在2022年2月初离我把Python和数据分析基础过完大概一个月。当时想往大数据方向走身边所有人的建议几乎一致先学PySpark。于是春节假期后的第一件事就是把Spark环境和Standalone集群跑起来。真正动手之后才发现Spark快速入门这几个字网上到处都是但能把Standalone集群从零搭好、提交第一个作业、再搞清楚作业在集群里到底怎么被调度执行的中间隔着一大堆教程不会明说的细节。这篇文章就把我这一天的学习过程完整复刻出来环境版本怎么搭配、集群参数怎么算、第一个PySpark作业怎么写、spark-submit怎么提交以及我在排查任务一直Pending时走过的弯路。适合刚学完Python准备入门大数据的人也适合已经在用local模式但一直没上过集群的读者。1. 学Spark之前先把Standalone模式的位置想清楚1.1 为什么我的第一个入口是PySpark而不是Scala在决定入门框架时我其实纠结过要不要直接学Scala。后来权衡下来选了PySpark核心原因很现实我日常写逻辑用的都是Python换到PySpark之后RDD、DataFrame这些数据结构对我来说只是换了一批新API而不是换一门语言。这里要说明一下Spark的核心执行引擎是JVM上的那一套PySpark只是借助一套进程间通信机制把Python写的算子翻译给JVM去执行所以底层调度、容错、内存管理这些概念用PySpark学习和用Scala学习是完全等价的。另外一点是PySpark本身在工业界早就不是简化玩具了。很多团队做离线特征、数据加工就是直接拿PySpark跑生产任务。也就是说我花时间学的这套东西后面真正工作也用得上不存在先学个简单的、以后还得推翻重学的问题。对当时只有Python基础、首门编译型语言都还没啃的我来说PySpark明显是投入产出比最高的选择。1.2 Local、Standalone、YARN、K8s四个模式要按这个顺序选Spark启动时有个--master参数它的值决定了你的任务跑在哪种资源调度环境里。最常见的四个值local[*]本地线程、spark://master:7077Standalone集群、yarnYARN资源管理器、k8s://Kubernetes集群。这四个不是平级的替代品学习顺序也很有讲究。local[*]只是单个JVM进程里用多线程模拟并行适合写代码和单机调试但它完全不涉及跨进程、跨机器的资源调度。Standalone是Spark自带的集群管理器不需要预先装Hadoop启动一个Master进程和若干Worker进程就能组成一个最小集群分布式调度、Executor分配这些核心概念全都有入门阶段用它学成本最低。YARN是工业界最常见的生产环境之一但它要依赖Hadoop生态组件多、排错链路长不适合拿来当第一个实验环境。Kubernetes更不用说容器编排那一套还没搞明白就贸然往上叠加很容易劝退。我的建议很明确先local模式把API写熟再上单机版Standalone集群把主从关系搞懂然后扩展到多节点最后有必要再接触YARN。标题里这个Standalone集群恰恰就是这条路上最值得认真走的一步。2. 环境准备Java、Python与Spark的版本三角关系2.1 版本组合怎么定为什么要用这个组合先说结论我2022年2月选定的组合是JDK 8、Python 3.8、Spark 3.2.1。这个组合不是随便拼的每个选择都有原因。组件版本选择原因JDK8Spark 3.2支持Java 8/11JDK 8生态资料最全、运行期坑最少Python3.8PySpark官方支持3.6-3.93.10刚发布不久第三方库兼容性还不稳Spark3.2.12022年1月发布的稳定版修掉了3.2.0的一些已知问题有个新手很容易忽略的事实即使你全程写Python机器上依然必须装JDK。原因是Spark的Driver和Executor本质上是JVM进程PySpark进程和JVM进程之间要通过通信机制交换数据和指令没有JVMPySpark根本跑不起来。所以我会Python就可以不用装Java这个想法趁早丢掉。选版本时也别一味的追新大数据生态里版本兼容性比新特性重要得多稳定组合才能少踩莫名其妙的坑。2.2 解压即用目录结构和两套命令的分工Spark的发行版是编译好的二进制包下载解压之后不需要额外编译就能用。但它的目录结构里有两套命令很多人一开始分不清bin目录下的spark-submit、pyspark、spark-shell是面向应用程序的sbin目录下的start-master.sh、start-workers.sh是面向集群本身的。这两套东西各自独立别搞混。jars目录是Spark运行需要的全部依赖配置都在conf目录下python目录里是PySpark的源码和打包好的Python包。我装好之后做的第一件事不是急着启动集群而是先看一眼spark-submit --version的输出。这一条命令能同时确认Spark版本和JVM版本是否正常如果这里都报错后面集群再折腾也没意义。很多入门教程第一步就让你直接start-master.sh其实跳过环境验证后面出了问题反而更难定位。2.3 环境变量与第一次启动验证环境变量方面核心是四件事JAVA_HOME、SPARK_HOME、PATH以及一个很多教程不强调的PYSPARK_PYTHON。前三个好理解最后一个是在集群模式下指定Worker节点用哪个Python解释器去执行Python代码的。如果你的Master和Worker不在同一台机器上几台机器上的Python路径不一致执行任务时就会莫名报错而报错信息往往还很隐蔽。所以从第一天起我就在所有节点的配置里统一写死了同一个Python路径。全部配好后我先启动pyspark用默认的local模式跑了一个最简单的RDD求和确认解释器、JVM、PySpark三方通信没问题。这一步的价值是建立基线如果你知道local模式是好的那之后再出问题就能确定问题出在集群配置上而不是环境本身。环境这关慢慢来急躁跳过验证后面排错的时间一定加倍还回来。3. Standalone集群搭建Master、Worker与资源数字3.1 先看懂spark://master:7077背后发生了什么Standalone模式下--master参数长得像spark://192.168.1.10:7077。这里面的7077是Master进程对外提供RPC服务的端口Driver通过它向Master注册Application、申请资源。注意别把它和8080混淆8080是Master的Web UI端口专门给人看的7077才是机器和机器之间通信用的。整个集群的工作流程大致是Worker进程启动后主动向Master注册自己当前有多少核、多少内存当Driver提交一个Application过来Master就根据这个Application申请的资源数在注册的Worker里挑出合适的节点为它在Worker上拉起Executor进程之后真正的计算在Executor里完成数据分片和任务分配则由Driver里的调度器负责。也就是说Master管谁有资源和怎么分配Driver管具体每个任务怎么切分。刚开始不用把每个概念都背下来只要知道这个主从关系和资源申请链路后面看日志和Web UI就顺了。3.2 配置文件逐行拆解spark-env.sh和workers集群配置都在conf目录下。首次使用要把spark-env.sh.template复制成spark-env.sh我在这份文件里写了四行核心配置SPARK_MASTER_HOST192.168.1.10 SPARK_WORKER_CORES3 SPARK_WORKER_MEMORY6g SPARK_WORKER_INSTANCES1SPARK_MASTER_HOST指定Master绑定到哪个IP多网卡机器一定要显式写否则可能绑定到奇怪地址上从别的机器就根本连不上。SPARK_WORKER_CORES和SPARK_WORKER_MEMORY表示每个Worker愿意贡献给Spark的核数和内存量这是集群资源的总池子。SPARK_WORKER_INSTANCES一般保持1就行只有你想在一台机器上跑多个Worker进程时才需要改。另外一个文件是conf目录下的workers文件。需要说明的是Spark 3.x里这个文件叫workers老版本叫slaves网上老教程混着叫别被搞晕。里面一行一个Worker节点的主机名或IP。我一开始就漏了维护这个文件结果start-workers.sh启动了个寂寞之后立刻补齐就好了。还有一个容易踩的配置是spark.local.dir它决定Executor写临时数据和shuffle文件的磁盘位置默认在/tmp下。如果/tmp空间很小大作业跑到一半会出现Task Lost我后来直接把它指到了数据盘上一个专门的目录一次就清净了。3.3 用一台4核8G机器示范资源怎么算资源参数不是拍脑袋填的我用一台4核8G的虚拟机演示一下计算逻辑。首先操作系统和系统进程本身要占资源不能把整台机器的CPU和内存全交给Spark。我的规矩是至少留1个核和大约2G内存给系统用。于是SPARK_WORKER_CORES填3SPARK_WORKER_MEMORY填6g。接着算应用侧的额度。当你用spark-submit提交作业时会指定--executor-memory假设我填5g。Executor在实际运行前除了堆内存本身Spark还要计算一部分额外开销默认大约是max(384MB, executor内存的10%)所以5g的Executor真实占用是5g加约0.5g合计约5.5g。这个数字必须小于Worker上报的6gExecutor才放得进去。如果我贪心填6g那真实占用就是6.6g超过Worker的6g容量应用就会永远停在等待资源的状态后面我会专门讲这个坑。所以一套合理的换算公式是executor-memory加max(384MB, 10% × executor-memory) 必须小于等于SPARK_WORKER_MEMORY。学习阶段我通常把Executor内存设为Worker内存的一半左右既留足余量又能一次多放几个Executor观察调度效果。这里给个具体例子Executor申请4g时开销约410MB总占用约4.4g放进6g的Worker绰绰有余如果Executor申请6g总占用6.6g放不进去。3.4 启动顺序、Web UI和进程检查启动顺序是先Master后Worker。第一台机器上执行sbin/start-master.sh然后执行sbin/start-workers.sh spark://192.168.1.10:7077。这里补充一句老教程里你可能还会看到start-slaves.sh和start-all.sh前者是Spark 2.x时代的叫法后者会把Master和Worker一次全启动但Hadoop也有一个同名的start-all.sh两套东西混在一起特别容易误操作所以我学习阶段一直用分开的脚本宁可多敲一条命令也不冒这个险。启动完之后的检查我固定做三件事一是用jps看进程正常应该看到Master和Worker两个Java进程二是访问http://192.168.1.10:8080Web UI里Workers表格会显示每个Worker的地址、核数、内存和状态三是打开logs目录下对应的.out日志文件确认没有异常堆栈。这三关都过了集群才算真正立起来了。4. 第一个PySpark作业词频统计的执行链4.1 RDD还是DataFrame学习期我建议先写一遍RDD现在PySpark教程很多一上来就教DataFrame这没错DataFrame有Schema、有优化器生产上基本都用它。但我还是建议学习期的第一个作业用RDD写一遍。原因是RDD把Spark的底层执行模型暴露得最直接惰性求值、窄依赖、宽依赖、shuffle、Stage划分这些概念靠DataFrame是看不出来的。而排查线上作业慢、任务卡住的时候你看Web UI里的Stage和Task粒度本质上还是在看RDD的执行模型。先把底层逻辑跑通再去看DataFrame的优化理解会扎实很多。等了解了RDD用DataFrame重写同一个作业作为对比这一步非常值得做两种API各写一遍对Spark的认知会立体很多。4.2 代码逐行拆解flatMap、map、reduceByKey词频统计是Spark版的Hello World完整代码长这样import sys from pyspark.sql import SparkSession spark SparkSession.builder.appName(WordCountLearn).getOrCreate() sc spark.sparkContext input_rdd sc.textFile(sys.argv[1]) words input_rdd.flatMap(lambda line: line.split()) pairs words.map(lambda word: (word, 1)) counts pairs.reduceByKey(lambda a, b: a b) counts.saveAsTextFile(sys.argv[2]) spark.stop()sc.textFile读入文件后得到一个RDD这时候数据还没真正读所有转换操作都是惰性的。flatMap把每一行按空格切词并把多行铺平成一个个单词map把每个单词变成(word, 1)这样的键值对reduceByKey把所有相同单词的计数两两相加。而真正触发计算的是最后一行saveAsTextFile它是个Action只有Action才会让Spark真正启动一个Job去执行前面所有惰性算子。执行过程中Spark会把整个Job按shuffle边界切成Stage。词频统计里reduceByKey触发了shuffle所以在这个作业里flatMap和map属于同一个StagereduceByKey和saveAsTextFile属于下一个Stage。每个Stage会被继续切成若干个Task分发给不同Executor并行执行。从Web UI的Stage页里你能清楚看到这两个Stage以及Task的数量这就是我前面说值得用RDD写一遍的原因。作为对比DataFrame版本连写起来都更接近SQLfrom pyspark.sql.functions import split, explode, lower lines_df spark.read.text(sys.argv[1]) words_df lines_df.select(explode(split(lower(lines_df[value]), r\W)).alias(word)) result words_df.groupBy(word).count() result.write.csv(sys.argv[2])同样一个统计逻辑DataFrame的写法更简洁、更结构化但底层优化细节被掩盖了。两种都写一遍你会明显感觉到Spark两个API家族的风格差异也更容易理解为什么现在团队里更偏爱DataFrame。4.3 从local[*]切到Standalone后发生了什么我先用spark-submit --master local[*]把词频统计跑通确认代码没问题再切到--master spark://192.168.1.10:7077。切换后最大的区别在于local模式下Driver和Executor都在同一个JVM进程里而Standalone client模式下Driver还在提交命令的这台机器上但Executor被分配到了Worker节点上。这个区别引发了一个我差点没发现的经典误区sc.textFile里如果写的是本地文件路径比如/data/input.txt在local模式没问题但在多节点集群里每个Executor会到自己所在机器上找这个路径。文件在Driver机器上并不代表它在所有Worker机器上都在。解决思路是让所有节点共享同一份输入数据要么用分布式存储要么放大家一起挂载的共享目录临时学习场景也可以在每台Worker上各放一份。输出路径也有同样的坑。saveAsTextFile如果写本地路径part文件会落在各Executor所在机器的本地磁盘上而不是回到Driver机器。我第一次跑完在Driver机器上找输出目录怎么都找不到后来才反应过来文件是被每个Executor分别写在各自节点上的。这个认知对理解分布式存储很重要凡是涉及文件路径的东西在集群模式下都默认指向Executor所在的机器而不是你敲命令的那台。5. spark-submit提交姿势与资源参数生效规则5.1 一条完整的提交命令拆给你看在掌握集群搭建和作业逻辑之后spark-submit就成了每天都要用的命令。我最常用的一条提交命令长这样spark-submit \ --master spark://192.168.1.10:7077 \ --deploy-mode client \ --name wordcount \ --executor-memory 4g \ --executor-cores 2 \ --total-executor-cores 4 \ --conf spark.sql.shuffle.partitions8 \ word_count.py /data/input.txt /data/output--executor-memory给每个Executor分配多少内存--executor-cores给每个Executor分配几个CPU核--total-executor-cores限制整个应用最多能用多少个核。最后那个--conf是给配置追加任意参数比如把shuffle的默认分区数从200改成8这个在做小数据实验时能省不少调度开销。学习阶段我的习惯是先在本地用local[*]跑一遍确认脚本正确再带着这份spark-submit命令去集群执行两边分开排查定位速度会快很多。5.2 Standalone模式下Executor是怎么排布的很多从YARN教程转过来的人第一反应是提交时写--num-executors来指定Executor数量。这里有个大坑在Standalone模式下--num-executors基本不生效真正决定Executor数量和排布的是cores和memory的组合。举个具体例子Worker上报的是3核6g如果我提交时指定--executor-cores 2、--executor-memory 4g那么这台Worker能放得下的Executor最多是1个因为2核的Executor已经占据了Worker的大部分核数虽然还剩下1核但内存已经被4g加开销占了大半。反过来如果我指定--executor-cores 1、--executor-memory 2g那这台上报3核6g的Worker理论上可以塞下两个Executor因为核和内存都够。Standalone调度器就是这么按Worker的空闲核数和内存逐台放置Executor的理解了这个逻辑你对为什么任务等了半天没跑起来的排查方向就会清晰很多。命令里还有一个--total-executor-cores它相当于给整个应用设了核数上限。我跑单机Standalone时习惯把它设成Worker核数的一半到三分之二既能看到多Executor并行又不会把整个Worker占死给后面的实验留出空间。5.3 client与cluster模式以及PySpark的特别限制--deploy-mode有两个选择client和cluster。client模式下Driver跑在提交命令的那台机器上日志直接打印在终端调试最方便这也是我学习阶段一直用的模式。cluster模式则是让集群自己拉一个进程当Driver日志不直接出现在你的终端。这里有一个PySpark初学者容易踩的专门限制在Standalone模式下cluster模式只对Java/Scala应用开放Python应用是不支持的。也就是说你拿PySpark脚本配合Standalone集群只能用client模式。不是命令敲错了而是这个组合在架构上就没开放那条路。如果真要用PySpark的cluster模式就得切换到YARN或者Kubernetes环境。知道这个限制能省下不少查文档的时间。6. 踩坑复盘三次现场还原与排查链路6.1 Worker反复注册失败hostname引发的连锁反应我第一天搭建多节点Standalone集群时出现过一个很典型的连环问题Master进程起来了Web UI也能访问但第二台机器上的Worker一遍遍尝试连接Master日志里反复出现连接失败Web UI里死活看不见这个Worker。我的排查链路是这样的先在第二台机器上ping Master的IP网络通再用nc或者类似工具测试7077端口端口也能通问题就不在网络不通而在主机名解析。原来我在Master机器上没配好主机名映射导致Worker拿着spark://某个hostname这个地址去连解析出来的却是回环地址自然连不上真正的Master。最后我把各机器的主机名和IP写进/etc/hosts确保所有节点都能通过hostname互相解析问题才彻底解决。那次之后我记住了集群里一切连接问题先查网络连通性再查端口最后一定记得查主机名解析顺序不能乱。6.2 应用一直PendingExecutor内存塞不进去另一个差点让我怀疑人生的现象是spark-submit提交后终端显示Application已经注册成功但接下来就是一直等在那里状态始终是等待资源。打开Master的Web UI能看到Application也能看到Worker就老老实实挂在那个集群里核数和内存都有剩余但Executor就是起不来。翻Master和Driver日志后我找到了真正原因Executor申请5g内存实际需要5.5g左右而Worker上报的内存只有6g扣除系统开销后其实已经不太够调度器一直在尝试放置却放不下。我用前面那套公式反复核算后把--executor-memory降到3g重新提交Application立刻进入Running状态。这个坑的通用排查思路是看到registered但pending优先怀疑资源放不进Worker用内存开销公式重新算一遍别急着怀疑代码逻辑。6.3 文件路径跑哪去了本地文件与输出目录的误区这个问题我前面提过但值得单独再说一次完整现象在单机local模式写完的word_count.py切到集群提交后输入文件明明还在Driver机器上放着任务却报文件找不到好不容易把输入文件复制到各Worker节点跑通了输出目录又在Driver机器上找不到。两个现象本质是同一件事集群模式下Executor所在的节点才是读写文件的主场。理解了这一点学习阶段的处理就很朴素了多节点集群实验优先用所有人都能访问的共享路径或者干脆在每个Worker节点都放一份输入文件输出直接写分布式存储或共享目录。如果你的集群只有一台机器Master和Worker同机那本地路径暂时还能凑合但心里必须清楚这只是碰巧在同一台机器上而已。集群从单节点扩展到多节点的那一刻这个坑一定会重新出现早理解早受益。6.4 学习阶段值得长期保留的检查命令清单经过这一天的折腾我给自己固定了一套下判断前先检查的命令清单每次集群状态不对就按这个顺序过一遍。首先是看进程jps能列出的Java进程Master、Worker、SparkSubmit都在就有基本的健康度然后是看端口7077是Master的RPC端口8080是Master Web UI4040是运行中应用自己的UI哪个端口不通就往哪个方向查接着是看资源free -g看系统内存还剩多少配合Master Web UI里Worker上报的资源数做对比最后是看日志Spark的日志都集中在logs目录Master、Worker、应用各自有独立的输出文件里面通常把根因写得比Web UI清楚得多。这套习惯坚持下来后面遇到的绝大多数集群问题都能在十分钟内定位到方向。对了我还养成了一个习惯每调通一个环境就把当时能正常运行的spark-submit完整命令连同配置文件一起复制进我的学习日志里下次重建环境直接照着抄。我自己的体会是大数据学习里最难的其实不是API用法而是出了问题之后能不能有一条清晰的排查路径可走。