Spark-Scala与PySpark在EMR上的DAG及运行时长差异问题

我把PySpark任务转成Scala任务在EMR上执行，两者的参数配置、输入数据、业务逻辑完全一致，但出现两个明显问题：

1. 任务运行时长不一样
2. Spark UI里生成的DAG结构有差异

附上的PySpark版本spark-submit命令 #

spark-submit --conf spark.sql.files.maxPartitionBytes=268435456 \
--master yarn --deploy-mode cluster  --conf spark.yarn.maxAppAttempts=1 \
--conf spark.sql.adaptive.enabled=true --conf spark.dynamicAllocation.enabled=false \
--conf spark.sql.parquet.filterPushdown=true \
--conf spark.sql.adaptive.coalescePartitions.enabled=true \
--conf spark.sql.adaptive.advisoryPartitionSizeInBytes=268435456 \
--conf spark.sql.adaptive.optimizeSkewsInRebalancePartitions.enabled=true \
--conf spark.sql.adaptive.rebalancePartitionsSmallPartitionFactor=.5 \
--conf spark.sql.adaptive.coalescePartitions.parallelismFirst=false \
--conf spark.sql.adaptive.coalescePartitions.initialPartitionNum=12000 \
--conf spark.sql.adaptive.localShuffleReader.enabled=true \
--conf spark.shuffle.io.connectionTimeout=8000 \
--conf spark.network.timeout=50000s  --conf spark.files.fetchTimeout=600s \
--conf spark.serializer=org.apache.spark.serializer.KryoSerializer --conf spark.kryoserializer.buffer.max=1g \
--conf spark.memory.storageFraction=0.05 --conf spark.memory.fraction=.8 \
--conf spark.shuffle.compress=true --conf spark.shuffle.spill.compress=true \
--conf spark.hadoop.fs.s3.multipart.th.fraction.parts.completed=0.99 \
--conf spark.sql.objectHashAggregate.sortBased.fallbackThreshold=4000000 \
--conf spark.reducer.maxReqsInFlight=12 \
--conf spark.shuffle.io.retryWait=60s \
--conf spark.shuffle.io.maxRetries=10 \
--conf spark.reducer.maxSizeInFlight=256m \
--conf spark.reducer.maxBlocksInFlightPerAddress=100 \
--conf spark.io.compression.codec=zstd \
--conf spark.shuffle.service.enabled=true \
--conf spark.io.compression.zstd.level=3 \
--conf spark.executor.cores=5 \
--conf spark.executor.instances=1200 \
--conf spark.executor.memory=34g --conf spark.driver.memory=60g --conf spark.executor.memoryOverhead=5g --conf spark.driver.memoryOverhead=4g \
--conf spark.hadoop.fs.s3a.fast.output.enabled=true \
--conf spark.executor.extraJavaOptions="-XX:+UseG1GC -XX:+UnlockDiagnosticVMOptions -XX:+G1SummarizeConcMark -XX:InitiatingHeapOccupancyPercent=35 -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:OnOutOfMemoryError='kill -9 %p' -Djavax.net.ssl.trustStore=/home/hadoop/.config/certs/InternalAndExternalTrustStore.jks" --conf spark.driver.extraJavaOptions="-XX:+UseG1GC " \
test.py

DAG截图 #

• PySpark执行的DAG：
• Scala执行的DAG：

可能的差异原因分析 #

1. 跨语言通信的额外开销 #

Scala是JVM原生语言，PySpark依赖Py4J实现Python与JVM进程间通信，数据在两种语言间的序列化/反序列化会产生额外开销——尤其是涉及UDF、复杂数据转换的环节，直接拉长了任务运行时间。同时PySpark的部分算子比Scala版本多一层封装，执行效率天然更低。

2. Catalyst优化器的解析差异 #

即便业务逻辑一致，PySpark和Scala的API被Catalyst优化器解析时，可能生成不同的执行计划：

• 某些PySpark的API写法（比如DataFrame的特定转换逻辑）可能被解析为不同的逻辑算子，导致DAG结构变化，进而影响任务并行度、shuffle策略。
• Catalyst对Scala代码的优化支持更原生，能触发更多深层优化规则（如精准的谓词下推、列裁剪），而PySpark的部分语法可能无法完全触发这些优化。

3. 数据读取与分区处理差异 #

从UI输出行数来看，两者数据读取后的分区数或数据分布可能存在差异：

• PySpark读取Parquet等数据源时，spark.sql.files.maxPartitionBytes的实际生效逻辑可能与Scala不同，导致初始分区数或数据倾斜情况有差异，影响后续任务执行时长。
• 自适应执行（Adaptive Execution）在PySpark和Scala中的触发时机、处理逻辑存在细微差别，导致DAG中合并分区、调整并行度的操作不同。

4. 序列化与内存管理差异 #

虽然配置了相同的Kryo序列化，但PySpark需要额外处理Python对象到JVM对象的转换：

• Python对象的序列化开销远高于Scala的JVM对象，即便使用Kryo也无法消除跨语言的序列化成本，会增加内存占用和GC压力，间接拖慢任务运行。
• PySpark的内存管理还要兼顾Executor中Python Worker的内存占用，而Scala任务仅需管理JVM内存，内存利用率更高。

5. 执行环境与依赖差异 #

• PySpark任务会额外加载Python环境相关依赖，Scala任务仅依赖JVM类库，这会导致Executor启动时间、资源初始化时间存在差异——尤其是在1200个Executor的大规模集群下，累计的启动时间差会很明显。
• EMR上的PySpark默认环境与Scala环境可能存在隐性差异（如JVM版本、Python版本的细节配置），这些差异会间接影响任务执行效率。

验证排查建议 #

• 对比执行计划：分别在PySpark和Scala中执行explain(true)，查看逻辑计划和物理计划的差异，定位Catalyst优化的区别。
• 监控性能指标：通过Spark UI的Executor Tab查看GC时间、Shuffle读写量、任务执行时间分布，对比两者的资源消耗情况。
• 拆解测试任务：把任务拆分为单个阶段（比如仅测试数据读取、仅测试某个转换算子），逐一验证每个环节的耗时差异，定位瓶颈点。
• 检查API等价性：确保PySpark和Scala的API调用完全等价，比如select、filter的写法是否触发了相同的底层逻辑。

内容的提问来源于stack exchange，提问作者user3858193