Spark中调用count()动作后map转换算子未触发执行如何解决？

问题根因 #

1. Spark的惰性求值机制：map属于转换算子，只有调用行动算子才会实际执行。你原有代码里recordDF.rdd.map(...)之后没有对这个RDD调用任何行动算子，后面的recordDF.count()是独立的DataFrame计数操作，和前面的RDD转换链路完全无关，所以map里的逻辑从头到尾都没被触发。
2. 调用collect()触发执行但OOM的原因：collect()会把所有Executor上的300万条数据全部拉取到Driver节点内存中再执行map逻辑，数据量远超Driver内存上限，自然会抛出内存溢出。

解决方案 #

你需要满足两个核心要求：一是触发RDD转换逻辑分布式运行，二是避免把全量数据拉到Driver端，同时还要规避S3客户端序列化问题。

核心实现逻辑 #

1. 把副作用操作（复制S3文件）放到分布式行动算子中执行，推荐直接用foreachPartition算子按分区处理，既可以减少S3客户端初始化开销，也能规避序列化报错。
2. 不要直接在算子内部使用Driver端初始化的S3Client，这类网络客户端默认不可序列化，直接传到Executor会抛出序列化异常，需要在每个分区内单独初始化。

修正后的代码示例 #

def copyFilesToArchive(recordDF:DataFrame,s3Util:S3Client):Unit ={
    if(s3Util !=null) {
      logger.info(".copyFilesToArchive() : Before Copying the Files to Archive and no.of RDD Partitions ={}",recordDF.rdd.partitions.length);
      val fileCount = recordDF.count()
      logger.info(".copyFilesToArchive() : No.of Files to Copy ={}",fileCount);
      // foreachPartition是行动算子，会直接触发执行，不需要额外调用其他action
      recordDF.rdd.foreachPartition(partition => {
        // 每个分区内单独初始化S3客户端，按需调整配置参数
        val innerS3Client = S3Client.builder().build()
        partition.foreach(row => {
          val key = row.getAs[String]("object_key")
          val bucketName = row.getAs[String]("bucket_name")
          val targetBucketName = row.getAs[String]("target_bucket_name")
          val targetKey = "archive/" + key
          val copyObjectRequest = new CopyObjectRequest(bucketName, key, targetBucketName,targetKey )
          logger.info(".copyFilesToArchive() : Copying the File from ["+key+"] to ["+targetKey+"]");
          innerS3Client.copyObject(copyObjectRequest)
        })
        // 用完关闭客户端避免连接泄漏
        innerS3Client.close()
      })
      logger.info(".copyFilesToArchive() : All files copied to Archive Folder successfully");
    } else{
      logger.info(".copyFilesToArchive() : Skipping Moving the Files as S3 Util is null");
    }
}

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