如何在本地Spark集群模式（Jupyter）中使用Delta Cache功能？

核心限制：Delta Cache的环境依赖 #

Delta Cache是Databricks Runtime专属的磁盘缓存优化，本地Jupyter+Spark环境默认不支持该特性的完整功能，仅通过设置spark.databricks.io.cache.enabled无法触发缓存，需要补充配置并适配本地环境。

必须补充的缓存配置 #

在cell1中添加以下Spark配置：

# 指定本地缓存存储目录（需确保目录存在且有读写权限）
spark.conf.set("spark.databricks.io.cache.dir", "/path/to/local/cache/dir")
# 设置缓存最大磁盘占用（例如10GB）
spark.conf.set("spark.databricks.io.cache.maxDiskUsage", "10g")
# 开启缓存调试日志（可选，用于验证缓存是否生效）
spark.conf.set("spark.databricks.io.cache.logLevel", "INFO")

触发缓存的正确操作 #

仅执行select * from my_table2+show()可能因数据量过小、执行计划未触发全表扫描导致缓存未生效，可修改cell3为：

import time
# 执行全表扫描并标记缓存
df = spark.sql("select * from my_table2").cache()
# 触发数据加载到缓存
df.count()

# 第一次查询耗时
s = time.time()
df.show()
print(f"第一次耗时: {time.time() - s}")

# 第二次查询验证缓存效果
s = time.time()
df.show()
print(f"第二次耗时: {time.time() - s}")

实现"仅下载新增数据"的补充方案 #

Delta Cache是缓存已有数据块，若要仅下载新增数据，需结合Delta Lake的增量读取特性：

• 首次读取时记录Delta表的版本号或时间戳
• 后续读取仅加载增量数据：

# 第一次读取时记录最新版本
df = spark.sql("select * from my_table2")
last_version = spark.sql("DESCRIBE HISTORY my_table2").select("version").first()[0]

# 后续读取增量数据
incremental_df = spark.sql(f"""
  SELECT * FROM my_table2
  VERSION AS OF {last_version + 1}
""")

额外注意事项 #

• 本地Spark环境需确保Delta Lake版本与官方文档匹配，避免兼容性问题
• 缓存目录需有足够磁盘空间，且Spark进程对该目录有读写权限
• Delta Cache仅缓存Parquet格式的数据块，若Delta表存在频繁小文件合并操作，缓存效率会受影响

内容的提问来源于stack exchange，提问作者user3595632