Spark(YARN)中如何追踪图像数据的工作节点分配与RDD分片？

一、Spark(YARN)作业执行核心流程 #

• 客户端提交作业到YARN ResourceManager，RM负责分配ApplicationMaster（AM）进程
• AM向RM申请计算容器，在容器内启动Executor进程
• Driver（嵌入AM中）将作业拆解为多个Stage，每个Stage对应一组可并行执行的Task
• AM将Task分发到各个Executor，Executor负责缓存数据、执行计算并向Driver返回结果
• 所有Task执行完成后，AM向RM上报作业完成，释放所有占用的容器资源

二、控制台查看RDD分片内容（针对图像数据） #

你用df.collect()得到的是合并后的全量数据，要查看每个分片的内容，别直接用glom().collect()，试试下面的方法：

方法1：在Executor节点打印分片信息 #

// 遍历每个分片，打印分片ID和前2条图像数据（避免打印全量导致日志爆炸）
df.rdd.mapPartitionsWithIndex((partitionId, dataIter) => {
  println(s"=== 分片ID: $partitionId ===")
  // 取分片内前2条数据打印，可根据需求调整数量
  dataIter.take(2).foreach(imgData => {
    // 图像数据是数组格式，打印前5个元素预览
    println(s"图像数据预览: ${imgData.take(5).mkString(", ")}")
  })
  dataIter // 保持RDD迭代器继续流转
}).count() // 触发作业执行

注意：这里的println输出在Executor节点的日志里，如果客户端控制台看不到，去YARN的Executor日志页面查看（Spark UI的Executor页面可直接跳转到日志）。

方法2：将分片数据收集到Driver打印 #

如果需要在客户端控制台看，用累加器或者小批量收集：

import org.apache.spark.util.CollectionAccumulator

// 创建累加器存储分片数据预览
val partitionAcc = sc.collectionAccumulator[(Int, Array[Double])]("PartitionData")

df.rdd.mapPartitionsWithIndex((partitionId, dataIter) => {
  // 每个分片取1条数据存入累加器
  dataIter.take(1).foreach(imgData => {
    partitionAcc.add((partitionId, imgData.take(10))) // 取前10个元素
  })
  dataIter
}).count()

// 在Driver打印累加器中的分片数据
partitionAcc.value.foreach { case (pid, imgArr) =>
  println(s"分片$pid 预览数据: ${imgArr.mkString(", ")}")
}

三、图像数据的分片规则 #

• 默认分片逻辑：Spark读取图像数据（比如用spark.read.format("image")）时，会根据底层存储的块大小（比如HDFS默认128MB）划分分片，一个分片对应一个或多个文件块。如果是本地文件，按文件大小和spark.sql.files.maxPartitionBytes（默认128MB）拆分。
• 自定义分片：如果默认分片不均，可手动调整：
  • 读取时设置参数：spark.read.option("maxPartitionBytes", "64MB").format("image").load("path")
  • 读取后用repartition(n)强制分成n个分片（会触发shuffle），或coalesce(n)合并分片（无shuffle）
• 注意：避免单分片数据过大（比如超大图像文件），会导致Task执行超时，建议提前拆分大文件。

四、df.rdd.glom().collect()无输出的原因与解决 #

• 核心原因：glom()会把整个分片的所有数据打包成一个数组，collect()又把所有分片数组拉到Driver，图像数据是高维数组，很容易导致Driver内存溢出（OOM）或作业超时，直接卡住没输出。
• 修复方案：
  1. 先采样再调试：只处理小部分数据，避免压垮Driver

     // 采样10%的数据，查看分片情况
     df.rdd.sample(withReplacement = false, fraction = 0.1)
       .glom()
       .collect()
       .zipWithIndex
       .foreach { case (partitionArr, idx) =>
         println(s"采样分片$idx 数据量: ${partitionArr.size}")
       }
     

  2. 增大Driver内存：在提交作业时设置--driver-memory 8g（YARN模式还要加--driver-memory-overhead 2g）
  3. 放弃glom()：改用mapPartitionsWithIndex逐个处理分片，避免一次性拉取全量数据到Driver

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