Scala 到 Pyspark

Question

我正在尝试在 Dstream 和静态 RDD 之间执行连接。

PySpark

  #Create static data
    ip_classification_rdd = sc.parallelize([('log_name','enrichment_success')])
    #Broadcast it to all nodes
    ip_classification_rdd_broadcast = sc.broadcast(ip_classification_rdd)
    #Join stream with static dataset on field log_name      
    joinedStream = kafkaStream.transform(lambda rdd: rdd.join(ip_classification_rdd[log_name]))

我得到了这个例外： "您似乎正在尝试广播一个 RDD 或从一个 "

引用一个 RDD

斯卡拉

但是，这里有人有同样的要求：How to join a DStream with a non-stream file?

这就是解决方案：

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

在 Pyspark 中这个等价物是什么？

Answer 1

您的代码需要进行一些更改：

• 您不能广播RDD：而是在底层“数据”上广播：
• 然后您可以使用value() 方法在闭包内获取广播变量

以下是您更新后的代码的近似值：

 #Create static data
    data = [('log_name','enrichment_success')])
    #Broadcast it to all nodes
    ip_classification_broadcast = sc.broadcast(data)
    #Join stream with static dataset on field log_name      
    joinedStream = kafkaStream.transform(lambda rdd:  \
        rdd.join(ip_classification_broadcast.value().get[1]))