我在我的项目中使用Spark Streaming(Spark V1.6.0)和HBase,HBase(HBase V1.1.2)配置在具有广播变量的执行程序之间传输 . 首先是Spark Streaming应用程序 works ,大约2天后,将出现异常 .
val hBaseContext: HBaseContext = new HBaseContext(sc, HBaseCock.hBaseConfiguration())
private def _materialDStream(dStream: DStream[(String, Int)], columnName: String, batchSize: Int) = {
hBaseContext.streamBulkIncrement[(String, Int)](
dStream,
hTableName,
(t) => {
val rowKey = t._1
val incVal = t._2
val increment = new Increment(Bytes.toBytes(rowKey))
increment.addColumn(Bytes.toBytes(hFamily), Bytes.toBytes(columnName), incVal)
increment
}, batchSize)
}
可以找到HBaseContext的整个源文件HBaseContext.scala,下面可以找到一些片段 .
经过几天的运行后,将出现异常,堆栈跟踪为:
Unable to getConfig from broadcast
16/02/01 10:08:10 ERROR Executor: Exception in task 3.0 in stage 187175.0 (TID 561527)
逻辑如下:
-
使用config(HBaseContext)创建HBaseContext并广播配置(如果指定了文件路径,则将配置保存到文件中)
-
在连接HBase之前,首先它将检查字段配置是否为空,如果是,则从指定文件中恢复它,或者如果没有指定文件路径,则从广播变量恢复它 .
从广播变量恢复配置时会出现问题,并且在“configBroadcast.value.value”中从广播中读取值时发生异常 .
我想如果master失败则Spark Streaming不会恢复广播变量,而getOrCreate()用于获取SparkStreaming实例 . 我对HBaseContext.scala源代码更感兴趣,该文件首选广播变量来恢复值 . 那么在Spark Streaming中使用广播的最佳做法是什么?我是否需要将它们存储在文件中,比如HDFS中的文件?
class HBaseContext(@transient sc: SparkContext, @transient config: Configuration, val tmpHdfsConfgFile: String = null) extends Serializable{
@transient var tmpHdfsConfiguration: Configuration = config
val broadcastedConf = sc.broadcast(new SerializableWritable(config))
if(tmpHdfsConfgFile != null && config != null){
// save config to file
}
private def getConf(configBroadcast: Broadcast[SerializableWritable[Configuration]]): Configuration = {
if (tmpHdfsConfiguration != null) {
tmpHdfsConfiguration
} else if (tmpHdfsConfgFile != null) {
// read config from file
tmpHdfsConfiguration
}
if (tmpHdfsConfiguration == null) {
try {
// Exception happens here!!!
tmpHdfsConfiguration = configBroadcast.value.value
tmpHdfsConfiguration
} catch {
case ex: Exception => {
println("Unable to getConfig from broadcast")
}
}
}
tmpHdfsConfiguration
}
}
1 回答
由于某种原因重新启动火花作业后,广播变量将被重置 . 或者在作业失败后将驱动程序与尝试重新关联 .
在流式传输作业的情况下,要使用广播变量,应该在创建StreamingContext之前从sprarkContext初始化广播 . 这将确保在流式传输开始时广播变量可用 .