在不引起序列化问题的情况下丰富 SparkContext

Question

我正在尝试使用 Spark 处理来自 HBase 表的数据。 This blog post 给出了一个示例，说明如何使用 NewHadoopAPI 从任何 Hadoop InputFormat 读取数据。

我做了什么

由于我需要多次执行此操作，因此我尝试使用隐式来丰富SparkContext，以便我可以从 HBase 中的一组给定列中获取 RDD。我写了以下助手：

trait HBaseReadSupport {
  implicit def toHBaseSC(sc: SparkContext) = new HBaseSC(sc)

  implicit def bytes2string(bytes: Array[Byte]) = new String(bytes)
}


final class HBaseSC(sc: SparkContext) extends Serializable {
  def extract[A](data: Map[String, List[String]], result: Result, interpret: Array[Byte] => A) =
    data map { case (cf, columns) =>
      val content = columns map { column =>
        val cell = result.getColumnLatestCell(cf.getBytes, column.getBytes)

        column -> interpret(CellUtil.cloneValue(cell))
      } toMap

      cf -> content
    }

  def makeConf(table: String) = {
    val conf = HBaseConfiguration.create()

    conf.setBoolean("hbase.cluster.distributed", true)
    conf.setInt("hbase.client.scanner.caching", 10000)
    conf.set(TableInputFormat.INPUT_TABLE, table)

    conf
  }

  def hbase[A](table: String, data: Map[String, List[String]])
    (interpret: Array[Byte] => A) =

    sc.newAPIHadoopRDD(makeConf(table), classOf[TableInputFormat],
      classOf[ImmutableBytesWritable], classOf[Result]) map { case (key, row) =>
        Bytes.toString(key.get) -> extract(data, row, interpret)
      }

}

可以这样使用

val rdd = sc.hbase[String](table, Map(
  "cf" -> List("col1", "col2")
))

在这种情况下，我们得到一个 RDD (String, Map[String, Map[String, String]])，其中第一个组件是行键，第二个是一个映射，其键是列族，值是映射，其键是列，其内容是单元格值。

失败的地方

不幸的是，我的工作似乎引用了sc，它本身在设计上是不可序列化的。当我运行这项工作时，我得到的是

Exception in thread "main" org.apache.spark.SparkException: Job aborted: Task not serializable: java.io.NotSerializableException: org.apache.spark.SparkContext
at org.apache.spark.scheduler.DAGScheduler$$anonfun$org$apache$spark$scheduler$DAGScheduler$$abortStage$1.apply(DAGScheduler.scala:1028)

我可以删除辅助类并在我的工作中使用相同的内联逻辑，并且一切运行正常。但我想得到一些可以重复使用的东西，而不是一遍又一遍地编写相同的样板。

顺便说一下，这个问题并不特定于隐式，即使使用sc 的函数也会出现同样的问题。

为了比较，以下读取 TSV 文件的助手（我知道它已损坏，因为它不支持引用等，没关系）似乎工作正常：

trait TsvReadSupport {
  implicit def toTsvRDD(sc: SparkContext) = new TsvRDD(sc)
}

final class TsvRDD(val sc: SparkContext) extends Serializable {
  def tsv(path: String, fields: Seq[String], separator: Char = '\t') = sc.textFile(path) map { line =>
    val contents = line.split(separator).toList

    (fields, contents).zipped.toMap
  }
}

如何封装逻辑以从 HBase 读取行，而不会无意中捕获 SparkContext？

Answer 1

只需将@transient 注释添加到sc 变量：

final class HBaseSC(@transient val sc: SparkContext) extends Serializable {
  ...
}

并确保sc 没有在extract 函数中使用，因为它不适用于工作人员。

如果需要从分布式计算中访问 Spark 上下文，可以使用rdd.context 函数：

val rdd = sc.newAPIHadoopRDD(...)
rdd map {
  case (k, v) => 
    val ctx = rdd.context
    ....
}