广播变量未显示在分区 Apache Spark 中答案

【问题标题】：Broadcast Variables not showing inside Partitions Apache Spark广播变量未显示在分区 Apache Spark 中
【发布时间】：2016-01-30 10:11:37
【问题描述】：

场景和问题：我想根据查找表值向 JSON 对象添加两个属性，并将 JSON 插入 Mongo DB。我有保存查找表的广播变量。但是，正如您在代码中看到的那样，我无法在 foreachPartition 中访问它。它没有给我任何错误，只是没有显示任何内容。另外，因为它我无法将 JSON 插入 Mongo DB。我找不到对这种行为的任何解释。非常感谢任何解释或解决方法以使其工作。

这是我的完整代码：

object ProcessMicroBatchStreams {
val calculateDistance = udf { 
 (lat: String, lon: String) =>      
 GeoHash.getDistance(lat.toDouble, lon.toDouble) }
 val DB_NAME = "IRT"
 val COLLECTION_NAME = "sensordata"
 val records = Array[String]()

def main(args: Array[String]): Unit = {
  if (args.length < 0) {
  System.err.println("Usage: ProcessMicroBatchStreams <master> <input_directory>")
  System.exit(1)
}
val conf = new SparkConf()
  .setMaster("local[*]")
  .setAppName(this.getClass.getCanonicalName)
  .set("spark.hadoop.validateOutputSpecs", "false")
/*.set("spark.executor.instances", "3")
.set("spark.executor.memory", "18g")
.set("spark.executor.cores", "9")
.set("spark.task.cpus", "1")
.set("spark.driver.memory", "10g")*/

val sc = new SparkContext(conf)
val ssc = new StreamingContext(sc, Seconds(60))
val sqc = new SQLContext(sc)
val gpsLookUpTable = MapInput.cacheMappingTables(sc, sqc).persist(StorageLevel.MEMORY_AND_DISK_SER_2)
val broadcastTable = sc.broadcast(gpsLookUpTable)


ssc.textFileStream("hdfs://localhost:9000/inputDirectory/")
  .foreachRDD { rdd =>
  //broadcastTable.value.show() // I can access broadcast value here
  if (!rdd.partitions.isEmpty) {
    val partitionedRDD = rdd.repartition(4)
    partitionedRDD.foreachPartition {
      partition =>
        println("Inside Partition")
        broadcastTable.value.show() // I cannot access broadcast value here
        partition.foreach {
          row =>
            val items = row.split("\n")
            items.foreach { item =>
              val mongoColl = MongoClient()(DB_NAME)(COLLECTION_NAME)
              val jsonObject = new JSONObject(item)
              val latitude = jsonObject.getDouble(Constants.LATITUDE)
              val longitude = jsonObject.getDouble(Constants.LONGITUDE)

              // The broadcast value is not being shown here
              // However, there is no error shown
              // I cannot insert the value into Mongo DB
              val selectedRow = broadcastTable.value
                .filter("geoCode LIKE '" + GeoHash.subString(latitude, longitude) + "%'")
                .withColumn("Distance", calculateDistance(col("Lat"), col("Lon")))
                .orderBy("Distance")
                .select(Constants.TRACK_KM, Constants.TRACK_NAME).take(1)
              if (selectedRow.length != 0) {
                jsonObject.put(Constants.TRACK_KM, selectedRow(0).get(0))
                jsonObject.put(Constants.TRACK_NAME, selectedRow(0).get(1))
              }
              else {
                jsonObject.put(Constants.TRACK_KM, "NULL")
                jsonObject.put(Constants.TRACK_NAME, "NULL")
              }
              val record = JSON.parse(jsonObject.toString()).asInstanceOf[DBObject]
              mongoColl.insert(record)
            }
        }
    }
  }
}
sys.addShutdownHook {
  ssc.stop(true, true)
}

ssc.start()
ssc.awaitTermination()
}
}

【问题讨论】：

MapInput从何而来，gpsLookUpTable的类型是什么？
我有一个文件 GPSLOOKUPTABLE.json，它是一个查找表。 MapInput 函数使用该表的 SparkSQL 返回 Dataframe。函数为：def cacheMappingTables(sc: SparkContext, sqlContext: SQLContext): DataFrame = { val allRows = sc.textFile("hdfs://localhost:9000/supportFiles/GeoHashLookUpTable") sqlContext.read.json(allRows).registerTempTable("GeoHashLookUpTable") val gpsLookUpTable = sqlContext.sql(" SELECT Lat, Lon, geoCode, TrackKM, TrackCode, TrackName, SubTrackCode, SubTrackName FROM GeoHashLookUpTable ") return gpsLookUpTable }
@Patrick McGloin 请提供您的见解
@Erik Schmiege 请说出你的见解
我仍然被困在这里。任何人都知道如何并行处理并访问广播变量？

标签： json mongodb scala apache-spark apache-spark-sql

【解决方案1】：

您似乎正在尝试广播 RDD。试试这样的：

broadCastVal = gpsLookUpTable.collect
broadCastTable = sc.broadcast(broadCastVal)

你应该能够得到你期望的值。

【讨论】：

我正在尝试对代码中的广播值val selectedRow = broadcastTable.value .filter("geoCode LIKE '" + GeoHash.subString(latitude, longitude) + "%'") .withColumn("Distance", calculateDistance(col("Lat"), col("Lon"))) .orderBy("Distance") .select(Constants.TRACK_KM, Constants.TRACK_NAME).take(1) 进行一些选择操作，并使用收集返回数组，但我无法做到。我想知道是什么导致了这个问题。在我重新分区之前，我能够使用相同的方式访问这些值。有什么想法吗？
我在 spark shell 上搞砸了，发现了这一点。当您使用 broadCastTable.value 时，它会返回一个内存地址，然后 scala 会计算出如何打印它。您可以通过将以下行添加到您的代码 println(broadCastTable.value) 来检查这一点正确的方向，您可能会拼凑一些可以解决您问题的东西
我不确定你是怎么做到的，但我无法访问广播变量的索引，正如你提到的println(broadCastTable.value[1])。首先，访问索引时出错，所以我将println(broadCastTable.value[1]) 更改为println(broadCastTable.value(1))，但问题是它不能将Int 值作为输入，所以我再次更改为println(broadCastTable.value("1"))，但结果是它试图访问列和传递的值是列名。即使我提供了列名，我仍然无法检索该值。还有什么建议吗？
我犯了一个愚蠢的错误。你肯定需要 () 而不是 [] 用于 scala。我可以传入 (1) 的原因是因为我使用了一个数组。我被你的问题难住了。我想你总是可以跳过广播部分并加入 rdds，而不是在广播表中查找值。不确定有多少数据或您的 sla 是什么，但这可能是一种替代方案。我会看这个，因为我很好奇现在会发生什么。

【解决方案2】：

我对此并不完全确定，但经过两次这样的遭遇后，我正在写这个答案。我可以广播 RDD，但我无法访问该值。如果我创建一个列表或 treeMap，我也可以广播和检索该值。我不确定为什么。虽然，我还没有发现任何地方写着我们不能广播 RDD。

【讨论】：