如何：Pyspark 数据帧持久使用和回读

Question

我对 pyspark 很陌生，我遇到以下错误：

Py4JJavaError: An error occurred while calling o517.showString. 我读到这是由于内存不足：
Caused by: java.lang.OutOfMemoryError: GC overhead limit exceeded
所以，我一直在读到这种情况的转机是使用df.persist()，然后再次阅读 persisted df，所以我想知道：

• 给定一个for 循环，我在其中执行一些.join 操作，我应该在循环内部还是在循环末尾使用.persist()？例如

  for col in columns:
     df_AA = df_AA.join(df_B, df_AA[col] == 'some_value', 'outer').persist()
  
  --> or <--
  
  for col in columns:
     df_AA = df_AA.join(df_B, df_AA[col] == 'some_value', 'outer')
  df_AA.persist()
  

• 完成后，我应该如何回读？ df_AA.unpersist()? sqlContext.read.some_thing(df_AA)?

我对此真的很陌生，所以请尽量解释清楚。

我正在使用 jupyter-notebooks(anaconda) 的本地机器（8GB 内存）上运行； Windows 7的;爪哇 8;蟒蛇3.7.1； pyspark v2.4.3

Answer 1

Spark 是惰性评估框架，因此，在您调用 action 之前，不会调用任何 转换，例如：join。

所以继续你所做的事情

from pyspark import StorageLevel
    for col in columns:
       df_AA = df_AA.join(df_B, df_AA[col] == 'some_value', 'outer')
    df_AA.persist(StorageLevel.MEMORY_AND_DISK)
    df_AA.show()

有多个持久化选项可用，因此选择 MEMORY_AND_DISK 会将内存中无法处理的数据溢出到 DISK。

GC 错误也可能是由于为 Spark 应用程序运行提供的 DRIVER 内存较少。