当我在闭包中使用扩展了serializable的case类或类/对象时,spark抛出任务not serializable。
object WriteToHbase extends Serializable {
def main(args: Array[String]) {
val csvRows: RDD[Array[String] = ...
val dateFormatter = DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss")
val usersRDD = csvRows.map(row => {
new UserTable(row(0), row(1), row(2), row(9), row(10), row(11))
})
processUsers(sc: SparkContext, usersRDD, dateFormatter)
})
}
def processUsers(sc: SparkContext, usersRDD: RDD[UserTable], dateFormatter: DateTimeFormatter): Unit = {
usersRDD.foreachPartition(part => {
val conf = HBaseConfiguration.create()
val table = new HTable(conf, tablename)
part.foreach(userRow => {
val id = userRow.id
val date1 = dateFormatter.parseDateTime(userRow.date1)
})
table.flushCommits()
table.close()
})
}
我的第一次尝试是使用case类:
case class UserTable(id: String, name: String, address: String, ...) extends Serializable
我的第二次尝试是使用类而不是case类:
class UserTable (val id: String, val name: String, val addtess: String, ...) extends Serializable {
}
我的第三次尝试是在类中使用伴随对象:
object UserTable extends Serializable {
def apply(id: String, name: String, address: String, ...) = new UserTable(id, name, address, ...)
}
2条答案
按热度按时间nkoocmlb1#
很可能函数“dosomething”是在类上定义的,它是不可序列化的。相反,将“dosomething”函数移到一个伴随对象上(例如,使其静止)。
lsmepo6l2#
它是dateformatter,我把它放在分区循环中,现在可以工作了。