python之数据框 pyspark 到 dict

我有这个数据框 path_df:

path_df.show() 
+---------------+-------------+----+ 
|FromComponentID|ToComponentID|Cost| 
+---------------+-------------+----+ 
|            160|          163|27.0| 
|            160|          183|27.0| 
|            161|          162|22.0| 
|            161|          170|31.0| 
|            162|          161|22.0| 
|            162|          167|24.0| 
|            163|          160|27.0| 
|            163|          164|27.0| 
|            164|          163|27.0| 
|            164|          165|35.0| 
|            165|          164|35.0| 
|            165|          166|33.0| 
|            166|          165|33.0| 
|            166|          167|31.0| 
|            167|          162|24.0| 
|            167|          166|31.0| 
|            167|          168|27.0| 
|            168|          167|27.0| 
|            168|          169|23.0| 
|            169|          168|23.0| 
+---------------+-------------+----+ 
only showing top 20 rows 

据此，我想做一个字典，如下: {FromComponentID:{ToComponentID:Cost}}

对于我当前的数据，它将是:

{160 : {163 : 27, 
        183 : 27}, 
 161 : {162 : 22, 
        170 : 31}, 
 162 : {161 : 22 
        167 : 24}, 
 ... 
 167 : {162 : 24, 
        166 : 31, 
        168 : 27} 
 168 : {167 : 27, 
        169 : 23}, 
 169 : {168 : 23} 
} 

我可以仅使用 PySpark 来实现吗？如何实现？或者最好提取我的数据并直接用 python 处理它们。

请您参考如下方法:

您可以使用数据帧转换和 udfs 完成所有这些。唯一有点烦人的是，因为从技术上讲你有两种不同类型的字典(一个是 key=integer 和 value=dictionary，另一个是 key=integer value=float)，你将不得不定义两个具有不同数据类型的 udf。这是一种可能的方法:

from pyspark.sql.functions import udf,collect_list,create_map 
from pyspark.sql.types import MapType,IntegerType,FloatType 
 
data = [[160,163,27.0],[160,183,27.0],[161,162,22.0], 
      [161,170,31.0],[162,161,22.0],[162,167,24.0], 
      [163,160,27.0],[163,164,27.0],[164,163,27.0], 
      [164,165,35.0],[165,164,35.0],[165,166,33.0], 
      [166,165,33.0],[166,167,31.0],[167,162,24.0], 
      [167,166,31.0],[167,168,27.0],[168,167,27.0], 
      [168,169,23.0],[169,168,23.0]] 
 
cols = ['FromComponentID','ToComponentID','Cost'] 
df = spark.createDataFrame(data,cols) 
 
combineMap = udf(lambda maps: {key:f[key] for f in maps for key in f}, 
             MapType(IntegerType(),FloatType())) 
 
combineDeepMap = udf(lambda maps: {key:f[key] for f in maps for key in f}, 
             MapType(IntegerType(),MapType(IntegerType(),FloatType()))) 
 
mapdf = df.groupBy('FromComponentID')\ 
.agg(collect_list(create_map('ToComponentID','Cost')).alias('maps'))\ 
.agg(combineDeepMap(collect_list(create_map('FromComponentID',combineMap('maps'))))) 
 
result_dict = mapdf.collect()[0][0] 

对于大型数据集，这应该比需要将数据收集到单个节点的解决方案提供一些性能提升。但是由于 spark 仍然需要序列化 ​​udf，因此与基于 rdd 的解决方案相比不会有很大的 yield 。

更新:

rdd 解决方案要紧凑得多，但在我看来，它并不那么干净。这是因为 pyspark 不会很容易地将大型词典存储为 rdds。解决方案是将其存储为元组的分布式列表，然后在将其收集到单个节点时将其转换为字典。这是一种可能的解决方案:

maprdd = df.rdd.groupBy(lambda x:x[0]).map(lambda x:(x[0],{y[1]:y[2] for y in x[1]})) 
result_dict = dict(maprdd.collect())  

同样，这应该比单节点上的纯 python 实现提供性能提升，它可能与数据帧实现没有太大区别，但我期望数据帧版本的性能更高。