是否可以从 Spark-streaming 检查点恢复广播值答案

【问题标题】：Is it possible to recover an broadcast value from Spark-streaming checkpoint是否可以从 Spark-streaming 检查点恢复广播值
【发布时间】：2018-09-21 20:59:07
【问题描述】：

我使用 hbase-spark 在我的 spark-streaming 项目中记录 pv/uv。然后当我杀死应用程序并重新启动它时，在检查点恢复时出现以下异常：

16/03/02 10:17:21 错误 HBaseContext：无法从广播中获取配置 java.lang.ClassCastException: [B 不能转换为 org.apache.spark.SerializableWritable 在 com.paitao.xmlife.contrib.hbase.HBaseContext.getConf(HBaseContext.scala:645) 在 com.paitao.xmlife.contrib.hbase.HBaseContext.com$paitao$xmlife$contrib$hbase$HBaseContext$$hbaseForeachPartition(HBaseContext.scala:627) 在 com.paitao.xmlife.contrib.hbase.HBaseContext$$anonfun$com$paitao$xmlife$contrib$hbase$HBaseContext$$bulkMutation$1.apply(HBaseContext.scala:457) 在 com.paitao.xmlife.contrib.hbase.HBaseContext$$anonfun$com$paitao$xmlife$contrib$hbase$HBaseContext$$bulkMutation$1.apply(HBaseContext.scala:457) 在 org.apache.spark.rdd.RDD$$anonfun$foreachPartition$1$$anonfun$apply$29.apply(RDD.scala:898) 在 org.apache.spark.rdd.RDD$$anonfun$foreachPartition$1$$anonfun$apply$29.apply(RDD.scala:898) 在 org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1839) 在 org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1839) 在 org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66) 在 org.apache.spark.scheduler.Task.run(Task.scala:88) 在 org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214) 在 java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) 在 java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) 在 java.lang.Thread.run(Thread.java:745)

我检查了 HBaseContext 的代码，它使用广播来存储 HBase 配置。

class HBaseContext(@transient sc: SparkContext,
               @transient config: Configuration,
               val tmpHdfsConfgFile: String = null) extends Serializable with Logging {

    @transient var credentials = SparkHadoopUtil.get.getCurrentUserCredentials()
    @transient var tmpHdfsConfiguration: Configuration = config
    @transient var appliedCredentials = false
    @transient val job = Job.getInstance(config)

    TableMapReduceUtil.initCredentials(job)
    // <-- broadcast for HBaseConfiguration here !!!
    var broadcastedConf = sc.broadcast(new SerializableWritable(config))
    var credentialsConf = sc.broadcast(new SerializableWritable(job.getCredentials()))
    ...

当检查点恢复时，它试图在它的 getConf 函数中访问这个广播值：

if (tmpHdfsConfiguration == null) {
  try {
    tmpHdfsConfiguration = configBroadcast.value.value
  } catch {
    case ex: Exception => logError("Unable to getConfig from broadcast", ex)
  }
}

然后引发异常。我的问题是：是否可以从 spark 应用程序中的检查点恢复广播值？我们还有其他解决方案来重新广播恢复后的价值吗？

感谢您的任何反馈！

【问题讨论】：

检查stackoverflow.com/questions/35124486/…

标签： spark-streaming

【解决方案1】：

目前，这是 Spark 的一个已知错误。贡献者一直在研究这个问题，但没有取得任何进展。

这是我的解决方法：我不是将数据加载到广播变量中并广播给所有执行器，而是让每个执行器将数据本身加载到单例对象中。

顺便说一句，请关注此问题以进行更改https://issues.apache.org/jira/browse/SPARK-5206

【讨论】：

谢谢，当我初始化单例对象时它可以工作
@Xuan Huy Pham & @He Bai .. 你能解释一下你的解决方案吗？如果我在 Map 函数中无法访问 Spark 上下文，执行程序如何将数据本身加载到单例对象中？
@Xuan.. 我以类似的方式进行了更改。创建该单例对象。我使用了 SparkConf。每次将任务从驱动程序序列化到工作人员时，我相信 SparkConf 会被序列化并检查单例对象是否已初始化。如果没有初始化，它会初始化否则没有动作。序列化 Sparkconf 是开销吗？？

【解决方案2】：

遵循以下方法

创建火花上下文。
初始化广播变量。
使用上述 spark 上下文创建带有检查点目录的流上下文并传递初始化的广播变量。

当流式作业开始时检查点目录中没有数据，它将初始化广播变量。

流重新启动时，它将从检查点目录中恢复广播变量。

【讨论】：