apache-spark - 如何使用非流文件加入 DStream？

Question

我想将 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？)。我看到的文档只列出了转换和连接，但没有显示静态文件是如何创建的。

关于如何广播地理数据然后将其加入每个流 RDD 的任何想法？

Answer 1

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)

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

val vdpJoinedGeo = goodIPsFltrBI.flatMap{ip => geoDataBC.value.get(ip).map(data=> (ip,data)}