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

Question

我已经看到并尝试了许多关于此问题的 existing StackOverflow 帖子，但没有一个有效。我想我的 JAVA 堆空间没有我的大型数据集预期的那么大，我的数据集包含 650 万行。我的 Linux 实例包含 4 核的 64GB 内存。根据suggestion，我需要修复我的代码，但我认为从 pyspark 数据帧制作字典的成本应该不会很高。如果有其他方法可以计算，请告诉我。

我只想从我的 pyspark 数据框制作一个 python 字典，这是我的 pyspark 数据框的内容，

property_sql_df.show() 显示，

+--------------+------------+--------------------+--------------------+
|            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 作为 列表 值的字典。

预期输出

{
  "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)

Answer 1

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

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

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

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 }

Answer 2

从链接的帖子中添加已接受的答案以供后代使用。答案通过利用write.json 方法解决了问题，并在此处防止向驱动程序收集过大的数据集：

https://stackoverflow.com/a/63111765/12378881

Answer 3

以下是如何使用您的数据制作示例 DataFrame：

data = [
    ("BOND-9129450", "90cb"),
    ("BOND-1742850", "d5c3"),
    ("BOND-3211356", "811f"),
    ("BOND-7630290", "d5c3"),
    ("BOND-7175508", "90cb"),
]
df = spark.createDataFrame(data, ["id", "hash_of_cc_pn_li"])

让我们在 Spark DataFrame 中聚合数据，以限制在驱动程序节点上收集的行数。我们将使用quinn 中定义的two_columns_to_dictionary 函数来创建字典。

agg_df = df.groupBy("hash_of_cc_pn_li").agg(F.max("hash_of_cc_pn_li").alias("hash"), F.collect_list("id").alias("id"))
res = quinn.two_columns_to_dictionary(agg_df, "hash", "id")
print(res) # => {'811f': ['BOND-3211356'], 'd5c3': ['BOND-1742850', 'BOND-7630290'], '90cb': ['BOND-9129450', 'BOND-7175508']}

这可能适用于相对较小的 650 万行数据集，但不适用于大型数据集。 “我认为从 pyspark 数据帧制作字典的成本应该不会很高”仅适用于非常小的数据帧。从 PySpark DataFrame 制作字典实际上非常昂贵。

PySpark 是一个集群计算框架，它得益于将数据分布在集群中的节点上。当您调用collect 时，所有数据都将移动到驱动程序节点，而工作程序节点无济于事。每当您尝试将过多数据移动到驱动程序节点时，都会出现 OutOfMemory 异常。

最好完全避免使用字典并找出解决问题的不同方法。好问题。

Answer 4

从 Spark-2.4 我们可以在这种情况下使用 groupBy,collect_list,map_from_arrays,to_json 内置函数。

Example:

df.show()
#+------------+-----------------+
#|          id| hash_of_cc_pn_li|
#+------------+-----------------+
#|BOND-9129450|90cb0946cf4139e12|
#|BOND-7175508|90cb0946cf4139e12|
#|BOND-1742850|d5c301f00e9966483|
#|BOND-7630290|d5c301f00e9966483|
#+------------+-----------------+
df.groupBy(col("hash_of_cc_pn_li")).\
agg(collect_list(col("id")).alias("id")).\
selectExpr("to_json(map_from_arrays(array(hash_of_cc_pn_li),array(id))) as output").\
show(10,False)
#+-----------------------------------------------------+
#|output                                               |
#+-----------------------------------------------------+
#|{"90cb0946cf4139e12":["BOND-9129450","BOND-7175508"]}|
#|{"d5c301f00e9966483":["BOND-1742850","BOND-7630290"]}|
#+-----------------------------------------------------+

要获得 one dict，请使用另一个带有 collect_list 的 agg。

df.groupBy(col("hash_of_cc_pn_li")).\
agg(collect_list(col("id")).alias("id")).\
agg(to_json(map_from_arrays(collect_list(col("hash_of_cc_pn_li")),collect_list(col("id")))).alias("output")).\
show(10,False)
#+---------------------------------------------------------------------------------------------------------+
#|output                                                                                                   |
#+---------------------------------------------------------------------------------------------------------+
#|{"90cb0946cf4139e12":["BOND-9129450","BOND-7175508"],"d5c301f00e9966483":["BOND-1742850","BOND-7630290"]}|
#+---------------------------------------------------------------------------------------------------------+