加入 pyspark Dataframes 时如何提高性能

Question

我有 2 个 pyspark Dataframess，第一个包含 ~500.000 rows，第二个包含 ~300.000 rows。我做了 2 个连接，在第二个连接中将从第二个数据帧（300.000 行）中逐个单元格地取出并将其与第一个数据帧（500.000 行）中的所有单元格进行比较。

所以，加入速度很慢。我在加入之前广播了数据帧。

测试 1：

df_join = df1.join(F.broadcast(df2), df1.String.contains(df2["search.subString"]), "left") 

这项工作花了很多时间才完成。

测试 2：

df_join = F.broadcast(df1).join(F.broadcast(df2), df1.String.contains(df2["search.subString"]), "left")

运行比上面第一个代码慢很多，所以性能很差。

我在加入前尝试 to cache the dataframes。

我用过：

df.cache() 每个数据帧。但是，成绩总是不好。

我尝试使用 persist in memory_only:

df.persist(MEMORY_ONLY) ==> NameError: global name 'MEMORY_ONLY' is not defined
df.persist(StorageLevel.MEMORY_ONLY) ==> NameError: global name 'StorageLevel' is not defined

如何将 Dataframe 保存在内存中？

你能给我一个提高性能的解决方案吗？

提前致谢。

Answer 1

使用

df=df.cache()

打印(df.count())

基本上需要调用一个action才能达到缓存的效果