spark sql insert hive non-partitioned table failure(java.io.NotSerializableException: org.apache.hadoop.mapreduce.Job)

[李斌松 <li...@gmail.com>]

*change sparkcontext.runJob，I'm not sure if it was due to my changes，*

*sparksql insert hive partition table success*

def runJob[T, U: ClassTag](rdd: RDD[T], func: (TaskContext,
Iterator[T]) => U): Array[U] = {
  val user = UserGroupInformation.getCurrentUser
  if(user == null){
    runJob(rdd, func, 0 until rdd.partitions.length)
  }else{
    var inner = new
DeliverLoginUser(UserGroupInformation.getCurrentUser.getUserName,
func)
    runJob(rdd, inner.jobFunc _, 0 until rdd.partitions.length)
  }
}


class DeliverLoginUser[T, U: ClassTag](userName: String, func:
(TaskContext, Iterator[T]) => U)
  extends Serializable with Logging {

  def jobFunc(taskContext: TaskContext, iterator: Iterator[T]): U = {
    val ugi = UserGroupInformation.createRemoteUser(userName)
    return ugi.doAs(new PrivilegedExceptionAction[U]() {
      override def run: U = {
        func(taskContext, iterator)
      }
    })
  }
}


Sorry exception information is not easily copied：
[image: 内嵌图片 1]

Re: spark sql insert hive non-partitioned table failure(java.io.NotSerializableException: org.apache.hadoop.mapreduce.Job)

[Ryan <ry...@gmail.com>]

It seems the func lambda includes a mapreduce.Job object which is not
serializable

On Mon, May 15, 2017 at 6:36 PM, 李斌松 <li...@gmail.com> wrote:

> *change sparkcontext.runJob，I'm not sure if it was due to my changes，*
>
> *sparksql insert hive partition table success*
>
> def runJob[T, U: ClassTag](rdd: RDD[T], func: (TaskContext, Iterator[T]) => U): Array[U] = {
>   val user = UserGroupInformation.getCurrentUser
>   if(user == null){
>     runJob(rdd, func, 0 until rdd.partitions.length)
>   }else{
>     var inner = new DeliverLoginUser(UserGroupInformation.getCurrentUser.getUserName, func)
>     runJob(rdd, inner.jobFunc _, 0 until rdd.partitions.length)
>   }
> }
>
>
> class DeliverLoginUser[T, U: ClassTag](userName: String, func: (TaskContext, Iterator[T]) => U)
>   extends Serializable with Logging {
>
>   def jobFunc(taskContext: TaskContext, iterator: Iterator[T]): U = {
>     val ugi = UserGroupInformation.createRemoteUser(userName)
>     return ugi.doAs(new PrivilegedExceptionAction[U]() {
>       override def run: U = {
>         func(taskContext, iterator)
>       }
>     })
>   }
> }
>
>
> Sorry exception information is not easily copied：
> [image: 内嵌图片 1]
>