我想使用Dataset.map函数来转换我的数据集中的行。示例如下:
val result = testRepository.readTable(db, tableName)
.map(testInstance.doSomeOperation)
.count()
其中testInstance
是一个扩展了java.io.Serializable
的类,但是testRepository
并没有继承这个类。代码会抛出以下错误:
Job aborted due to stage failure.
Caused by: NotSerializableException: TestRepository
问题
我理解为什么 testInstance.doSomeOperation
需要序列化,因为它在 map 内部并将分发到 Spark worker。但是为什么需要对 testRepository
进行序列化呢?我不明白这对于 map 来说是必要的。将其定义更改为 class TestRepository extends java.io.Serializable
可以解决此问题,但在项目的更大上下文中并不理想。
是否有一种方法可以在不使 TestRepository 序列化的情况下使其正常工作,或者为什么需要进行序列化?
最小工作示例
以下是包含两个类代码的完整示例,可重现 NotSerializableException:
import org.apache.spark.sql._
import org.apache.spark.sql.functions._
case class MyTableSchema(id: String, key: String, value: Double)
val db = "temp_autodelete"
val tableName = "serialization_test"
class TestRepository extends java.io.Serializable {
def readTable(database: String, tableName: String): Dataset[MyTableSchema] = {
spark.table(f"$database.$tableName")
.as[MyTableSchema]
}
}
val testRepository = new TestRepository()
class TestClass() extends java.io.Serializable {
def doSomeOperation(row: MyTableSchema): MyTableSchema = {
row
}
}
val testInstance = new TestClass()
val result = testRepository.readTable(db, tableName)
.map(testInstance.doSomeOperation)
.count()
object TestRepository
。 - botchniaque.map()
的代码需要序列化吗? - RvdV