我看过并尝试了许多关于此问题的现有StackOverflow帖子,但都没有起作用。我猜我的JAVA堆空间对于我的大数据集来说不够大,我的数据集包含650万行。我的Linux实例具有64GB RAM和4个核心。根据这个建议,我需要修复我的代码,但我认为从pyspark dataframe创建字典不应该很昂贵。请告诉我是否还有其他计算方法。
我只想从我的pyspark dataframe中创建一个Python字典,这是我的pyspark dataframe的内容,
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作为键,id作为列表值来创建一个字典。
期望输出
{
"90cb0946cf4139e12": ["BOND-9129450", "BOND-7175508"]
"d5c301f00e9966483": ["BOND-1742850","BOND-7630290"]
}
迄今为止我尝试过的方法:
方法1:导致java.lang.OutOfMemoryError:Java堆空间不足
%%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]
方法2:由于缺少pyspark的本地OFFSET而无法工作。
%%time
i = 0
limit = 1000000
for offset in range(0, total_record,limit):
i = i + 1
if i != 1:
offset = offset + 1
duplicate_property_list = {}
duplicate_properties = {}
# Preparing dataframe
url = '''select id, hash_of_cc_pn_li from properties_df LIMIT {} OFFSET {}'''.format(limit,offset)
properties_sql_df = spark.sql(url)
# Grouping dataset
rows = properties_sql_df.groupBy("hash_of_cc_pn_li").agg(F.collect_set("id").alias("ids")).collect()
duplicate_property_list = { row.hash_of_cc_pn_li: row.ids for row in rows }
# Filter a dictionary to keep elements only where duplicate cound
duplicate_properties = filterTheDict(duplicate_property_list, lambda elem : len(elem[1]) >=2)
# Writing to file
with open('duplicate_detected/duplicate_property_list_all_'+str(i)+'.json', 'w') as fp:
json.dump(duplicate_property_list, fp)
现在在控制台上看到的信息:
java.lang.OutOfMemoryError: Java heap space
并在Jupyter笔记本输出中显示此错误。
ERROR:py4j.java_gateway:An error occurred while trying to connect to the Java server (127.0.0.1:33097)
这是我在这里提出的后续问题: 从Pyspark数据框创建字典,显示OutOfMemoryError:Java堆空间
property_sql_df
之前/之后的任何转换。 - ernest_k