我在Spark / Scala REPL上尝试一个简单的代码并得到此错误 . 如何解决这个问题 . 我想使用RDD.saveAsNewAPIHadoopDataset（conf）将RDD保存到HBase
import org.apache.hadoop.hbase.{HBaseConfiguration, HTableDescriptor}
import org.apache.hadoop.hbase.client.HBaseAdmin
import org.apache.hadoop.hbase.mapreduce.TableInputFormat
import org.apache.hadoop.hbase.mapreduce.TableOutputFormat
import org.apache.hadoop.mapreduce.OutputFormat
import org.apache.hadoop.hbase.client.Mutation
import org.apache.hadoop.fs.Path
import org.apache.hadoop.hbase.HColumnDescriptor
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.conf.Configuration
import scala.collection.mutable
import java.io.{IOException, File, ByteArrayOutputStream}
import org.apache.hadoop.hbase.util.Bytes
import org.apache.spark.HashPartitioner
val tableName = "test"
val cfIndex = "cf".getBytes()
val colIndexId = "c1".getBytes()
val RDD = sc.parallelize(List(("1","2"),("1","2"),("1","3"), ("3","3")),2).repartition(2).mapPartitions { part =>
val tableName = "test"
val cfIndex = "cf".getBytes()
val colIndexId = "c01".getBytes()
part.map { case(k,v) =>
val put = new Put(k.getBytes())
put.add(cfIndex, colIndexId, v.getBytes())
(k, put)
}
}
错误TaskSetManager：阶段5.0（TID 17）中的任务0.0具有不可序列化的结果：org.apache.hadoop.hbase.client.Put
1 回答
Put不可序列化 - 因为它在你的map闭包中它将在worker节点上执行，因此需要序列化并从驱动程序发送到worker节点 . 由于它是不可序列化的，因此无法导致异常 .