Databricks Spark Streaming NoClassDefFoundError及foreachPartition报错解决咨询

错误原因 #

• 你在Notebook中直接定义的com.databricks.s3get包代码默认只会在Driver端存在，不会自动分发到Executor节点，Executor加载类时找不到对应类定义。
• 单例对象fetchS3Object的初始化逻辑包含dbutils.secrets.get调用和S3客户端初始化：dbutils默认仅在Driver端可用，且S3客户端本身不可序列化，在Driver端初始化后无法发送到Executor执行。
• 代码存在笔误：你调用的是writeS3Object.getS3Object(dataframe)，但定义的单例对象名为fetchS3Object，类名不匹配。

解决步骤 #

1. 将com.databricks.s3get的代码单独打成Jar包，上传到Databricks集群的库依赖中，确保所有Executor节点都能加载到对应类。
2. 移除单例对象中全局的S3客户端初始化逻辑和dbutils调用，将S3客户端初始化逻辑放到Executor端执行的代码中（比如每个分区初始化一次）。AK/SK可以在Driver端通过dbutils读取后，广播到Executor使用。
3. 修正调用笔误，将writeS3Object.getS3Object(dataframe)改为fetchS3Object.getS3Object(dataframe)。

错误原因 #

• 代码存在未定义变量：你调用client.getObject("s3bucketname", objectKey)时，objectKey变量没有定义，编译器类型推断出错，将partition识别为Object类型，因此找不到foreach方法。
• provides3Client方法中调用了dbutils.secrets.get，该方法只能在Driver端执行，直接在foreachPartition中调用会导致Executor端执行失败，同时也会引发序列化问题。
• 写入文件时没有提前创建父目录，会触发文件路径不存在的异常。

解决步骤 #

1. 修正未定义变量问题，将objectKey替换为你实际的S3对象名，比如和之前逻辑一致的s"${s3ObjectName}_.json"。
2. 调整密钥读取逻辑：在Driver端先通过dbutils读取AK/SK，再作为参数传入foreachPartition中，不要在Executor端调用dbutils。
3. 增加目录创建逻辑，写入文件前先调用new File(filePath).getParentFile.mkdirs()创建父目录。
4. 增加资源异常处理，用try-finally包裹流、文件写入器的关闭逻辑，避免资源泄漏。
5. 确保已经导入必要的依赖：

import org.apache.spark.sql._
import com.amazonaws.services.s3._
import org.apache.commons.io.IOUtils
import java.io._

修正后的参考代码：

// Driver端读取密钥
val accessKey = dbutils.secrets.get(scope = "My_Scope", key = "AccessKey-ID")
val secretKey = dbutils.secrets.get(scope = "My_Scope", key = "AccessKey-Secret")

dataframe.foreachPartition(partition => {
  // 每个分区初始化一次S3客户端
  val creds = new BasicAWSCredentials(accessKey, secretKey)
  val clientRegion: Regions = Regions.US_EAST_1
  val client: AmazonS3 = AmazonS3ClientBuilder.standard()
    .withRegion(clientRegion)
    .withCredentials(new AWSStaticCredentialsProvider(creds))
    .build()
  
  partition.foreach(row => {
    val s3ObjectName = row.getString(0)
    val batchname = row.getString(1)
    val objectKey = s"${s3ObjectName}_.json"
    var inputS3Stream: InputStream = null
    var bw: BufferedWriter = null
    try {
      inputS3Stream = client.getObject("s3bucketname", objectKey).getObjectContent
      val inputS3String = IOUtils.toString(inputS3Stream, "UTF-8")
      val filePath = s"/dbfs/mnt/pp-telematics-working-5/Sensors/s3response/test/${batchname}/${s3ObjectName}"
      // 提前创建目录
      new File(filePath).getParentFile.mkdirs()
      bw = new BufferedWriter(new FileWriter(new File(filePath)))
      bw.write(inputS3String)
    } finally {
      if(inputS3Stream != null) inputS3Stream.close()
      if(bw != null) bw.close()
    }
  })
})

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