从 Pyspark 数据帧创建字典显示 OutOfMemoryError: Java 堆空间

我见过并尝试过很多existing https://stackoverflow.com/questions/37335/how-to-deal-with-java-lang-outofmemoryerror-java-heap-space-errorStackOverflow 发布了有关此问题的帖子，但没有任何效果。我猜我的 JAVA 堆空间没有我的大型数据集预期的那么大，我的数据集包含 650 万行。我的 Linux 实例包含 4 核 64GB RAM。按照这个建议 https://stackoverflow.com/questions/37335/how-to-deal-with-java-lang-outofmemoryerror-java-heap-space-error我需要修复我的代码，但我认为从 pyspark 数据帧制作字典应该不会很昂贵。如果有其他计算方法，请告诉我。

我只想从我的 pyspark 数据帧创建一个 python 字典，这是我的 pyspark 数据帧的内容，

property_sql_df.show() shows,

+--------------+------------+--------------------+--------------------+
|            id|country_code|       name|          hash_of_cc_pn_li|
+--------------+------------+--------------------+--------------------+
|  BOND-9129450|          US|Scotron Home w/Ga...|90cb0946cf4139e12...|
|  BOND-1742850|          US|Sited in the Mead...|d5c301f00e9966483...|
|  BOND-3211356|          US|NEW LISTING - Com...|811fa26e240d726ec...|
|  BOND-7630290|          US|EC277- 9 Bedroom ...|d5c301f00e9966483...|
|  BOND-7175508|          US|East Hampton Retr...|90cb0946cf4139e12...|
+--------------+------------+--------------------+--------------------+

我想要的是用 hash_of_cc_pn_li 制作一本字典key和 id 作为a list value.

预期输出

{
  "90cb0946cf4139e12": ["BOND-9129450", "BOND-7175508"]
  "d5c301f00e9966483": ["BOND-1742850","BOND-7630290"]
}

到目前为止我所尝试过的

%%time
duplicate_property_list = {}
for ind in property_sql_df.collect(): 
     hashed_value = ind.hash_of_cc_pn_li
     property_id = ind.id
     if hashed_value in duplicate_property_list:
         duplicate_property_list[hashed_value].append(property_id) 
     else:
         duplicate_property_list[hashed_value] = [property_id] 

我现在在控制台上得到的内容：

java.lang.OutOfMemoryError：Java堆空间

并显示此错误Jupyter 笔记本输出

ERROR:py4j.java_gateway:An error occurred while trying to connect to the Java server (127.0.0.1:33097)

从 pyspark 数据帧制作字典应该不会很昂贵

就运行时而言确实如此，但这很容易占用大量空间。特别是如果你正在做property_sql_df.collect()，此时您将整个数据帧加载到驱动程序内存中。在 650 万行中，如果每行有 10KB 或 10K 个字符，那么您就已经达到 65GB，而且我们甚至还没有访问字典。

首先，您可以只收集您需要的列（例如，不收集name）。其次，您可以在 Spark 中进行上游聚合，这将节省一些空间，具体取决于有多少id是否有每hash_of_cc_pn_li:

rows = property_sql_df.groupBy("hash_of_cc_pn_li") \
  .agg(collect_set("id").alias("ids")) \
  .collect()

duplicate_property_list = { row.hash_of_cc_pn_li: row.ids for row in rows }