当尝试在Spark流中使用持久化表时，空指针异常

Question

我在开始创建"gpsLookUpTable"并持久化它，这样我就不需要一遍又一遍地拉它来做映射。然而，当我试图在foreach中访问它时，我得到空指针异常。任何帮助都很感激，谢谢。

下面是代码片段:

def main(args: Array[String]): Unit = { 
val conf = new SparkConf() ... 
val sc = new SparkContext(conf) 
val ssc = new StreamingContext(sc, Seconds(20)) 
val sqc = new SQLContext(sc) 
//////Trying to cache table here to use it below 
val gpsLookUpTable = MapInput.cacheMappingTables(sc, sqc).persist(StorageLevel.MEMORY_AND_DISK_SER_2) 
//sc.broadcast(gpsLookUpTable) 
ssc.textFileStream("hdfs://localhost:9000/inputDirectory/") 
.foreachRDD { rdd => 
if (!rdd.partitions.isEmpty) { 
val allRows = sc.textFile("hdfs://localhost:9000/supportFiles/GeoHashLookUpTable") 
sqc.read.json(allRows).registerTempTable("GeoHashLookUpTable") 
val header = rdd.first().split(",") 
val rowsWithoutHeader = Utils.dropHeader(rdd) 
rowsWithoutHeader.foreach { row => 
val singleRowArray = row.split(",") 
singleRowArray.foreach(println) 
(header, singleRowArray).zipped 
.foreach { (x, y) => 
///Trying to access persisted table but getting null pointer exception 
val selectedRow = gpsLookUpTable 
.filter("geoCode LIKE '" + GeoHash.subString(lattitude, longitude) + "%'") 
.withColumn("Distance", calculateDistance(col("Lat"), col("Lon"))) 
.orderBy("Distance") 
.select("TrackKM", "TrackName").take(1) 
if (selectedRow.length != 0) { 
// do something
} 
else { 
// do something
} 
} 
} }}

Answer 1

我假设您在集群中运行;您的foreach将作为闭包在其他节点上运行。空指针会被引发，因为闭包运行在没有初始化gpsLookUpTable的节点上。你显然试图在

中广播gpsLookUpTable

//sc.broadcast(gpsLookUpTable) 

但这需要绑定到一个变量，基本上是这样的:

val tableBC = sc.broadcast(gpsLookUpTable) 

在foreach中，您将替换为:

foreach { (x, y) => 
///Trying to access persisted table but getting null pointer exception 
val selectedRow = gpsLookUpTable 

与这个:

foreach { (x, y) => 
///Trying to access persisted table but getting null pointer exception 
val selectedRow = tableBC.value