4

我想将 DStream 中的每个 RDD 与非流式、不变的参考文件一起加入。这是我的代码：

val sparkConf = new SparkConf().setAppName("LogCounter") 
val ssc =  new StreamingContext(sparkConf, Seconds(2)) 

val sc = new SparkContext() 
val geoData = sc.textFile("data/geoRegion.csv") 
            .map(_.split(',')) 
            .map(line => (line(0), (line(1),line(2),line(3),line(4)))) 

val topicMap = topics.split(",").map((_,numThreads.toInt)).toMap 
val lines = KafkaUtils.createStream(ssc, zkQuorum, group, topicMap).map(_._2) 

val goodIPsFltrBI = lines.filter(...).map(...).filter(...) // details removed for brevity 
val vdpJoinedGeo = goodIPsFltrBI.transform(rdd =>rdd.join(geoData))

我收到很多很多错误，最常见的是：

14/11/19 19:58:23 WARN TaskSetManager: Loss was due to java.io.FileNotFoundException
java.io.FileNotFoundException: http://10.102.71.92:40764/broadcast_1

我认为我应该广播 geoData 而不是在每个任务中读取它（它是一个 100MB 的文件），但我不确定将第一次初始化 geoData 的代码放在哪里。

另外我不确定 geoData 是否定义正确（也许它应该使用 ssc 而不是 sc？）。我看到的文档只列出了转换和连接，但没有显示静态文件是如何创建的。

关于如何广播 geoData 然后将其加入每个流式 RDD 的任何想法？

4

1 回答 1

4
  • FileNotFound 异常：

geoData textFile 从提供的位置（“data/geroRegion.csv”）加载到所有工作人员上。很可能该文件仅在驱动程序中可用，因此工作人员无法加载它，从而引发文件未找到异常。

  • 广播变量：

广播变量在驱动程序上定义，并通过解开广播容器以获取内容在工作人员上使用。这意味着广播变量包含的数据应该在定义作业之前由驱动程序加载。

在这种情况下，这可能会解决两个问题：假设 geoData.csv 文件位于驱动程序节点中，它将允许在驱动程序上正确加载此数据并有效地分布在集群中。

在上面的代码中，将 geoData 加载替换为本地文件读取版本：

val geoData = Source.fromFile("data/geoRegion.csv").getLines 
            .map(_.split(',')) 
            .map(line => (line(0), (line(1),line(2),line(3),line(4)))).toMap 

val geoDataBC = sc.broadcast(geoData)

要使用它，您可以在闭包中访问广播内容。请注意，您将可以访问先前包装在广播变量中的地图：它是一个简单的对象，而不是 RDD，因此在这种情况下，您不能使用它join来合并两个数据集。您可以改用 flatMap ：

val vdpJoinedGeo = goodIPsFltrBI.flatMap{ip => geoDataBC.value.get(ip).map(data=> (ip,data)}
于 2014-11-20T09:57:34.047 回答