MEMORY_AND_DISK_DESER 如何用于 PySpark DataFrame？

Question

有人可以解释 PySpark DataFrame 缓存/持久化的默认

MEMORY_AND_DISK_DESER

存储级别的行为吗？

（以前DataFrame的默认存储级别好像是来自

MEMORY_AND_DISK

？）

我的理解是，Python 对象是（或曾经）序列化存储的，但想知道情况是否仍然如此。也许反序列化现在是默认的，至少对于内存中缓存的 Python 对象来说是这样？

（让事情变得复杂的是，我认为 RDD 的默认存储级别是

MEMORY_ONLY

，这希望超出了这个问题的范围！）

最新 Spark 3.5.1 文档（2024 年 4 月 18 日）的两次引用：

“在 Python 中，存储的对象将始终使用 Pickle 进行序列化 库，所以选择序列化级别并不重要”

https://spark.apache.org/docs/latest/rdd-programming-guide.html#rdd-persistence

“默认存储级别已更改为 MEMORY_AND_DISK_DESER 与 3.0 中的 Scala 相匹配。”

https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.cache.html

感谢您的帮助！

Answer 1

我用 Databricks PySpark 进行了一个小实验，似乎默认的

MEMORY_AND_DISK_DESER

的行为有所不同，并且缓存在内存中反序列化。

测试结果，以及每个缓存的 DataFrame 的内存大小：

• MEMORY_AND_DISK_DESER

  ：4.1Mb - 磁盘内存反序列化 1x 复制

• MEMORY_AND_DISK

  ：2.4Mb - 磁盘内存序列化 1x 复制

但不确定这些存储级别如何影响磁盘序列化。

步骤

1. 在 Databricks 笔记本单元中运行下面的代码（仅供参考：我使用了 Runtime 12.2）
2. 打开关联作业集群的 Databricks Spark UI“存储”选项卡
3. 比较两个缓存数据集内存中的大小属性

%python

from pyspark import StorageLevel
from pyspark.sql.functions import concat_ws, col, lit, rand, round

randy = rand(seed=0)

def create_df(storage_level, salt):
    df = spark.range(1, 1000000 + 1).withColumn("item_name", concat_ws(" ", lit(salt), round(randy * 1000)))
    df.persist(storage_level).count()
    return df

df1 = create_df(StorageLevel.MEMORY_AND_DISK_DESER, "df1")
df2 = create_df(StorageLevel.MEMORY_AND_DISK, "df2")

注意：每个 DataFrame 中使用相同大小的盐值以确保唯一性，以防止 Databricks 默默地将 df1 重新用作 df2。