如何缓存Spark会话终止后仍保留的DataFrame？解决数据库连接失败问题

先澄清你的误解 #

• Spark的df.persist(StorageLevel.DISK_ONLY)是会话级缓存，Spark作业结束（会话终止）后缓存就会被清理，完全不适合你这种每10分钟启动新Spark会话的场景，这个方案直接排除。

适合你场景的缓存方案 #

结合Azure生态，推荐以下几种实用方案：

1. 将数据写入Azure存储（ADLS Gen2/BLOB Storage），每日刷新

这是最适配你场景的轻量方案：

• 单独创建一个每日执行的ADF管道，负责从Azure SQL读取三张表，以Parquet格式写入ADLS Gen2或BLOB Storage（列式存储对小数据也高效，占用空间小）。
• 原有的10分钟执行的Spark作业，直接从存储读取Parquet文件，不再连接Azure SQL。
• 代码示例（Python）：

  # 每日刷新作业：写入存储
  df = spark.read.format("jdbc").options(**jdbc_config).load("table_name")
  df.write.mode("overwrite").parquet("abfss://container@yourstorage.dfs.core.windows.net/cached_tables/table_name")
  
  # 10分钟作业：读取缓存
  cached_df = spark.read.parquet("abfss://container@yourstorage.dfs.core.windows.net/cached_tables/table_name")
  

• 优势：成本极低，Azure原生集成，无需额外服务，完全解耦数据读取和业务作业，避免数据库连接依赖。

2. 使用Hive Metastore（含Azure Synapse内置Hive）

Hive并非只适用于大数据场景，小数据场景下用它做元数据管理非常方便：

• 每日执行的作业将数据写入Hive表（支持自动管理表结构）：

  df.write.mode("overwrite").saveAsTable("your_db.cached_table_name")
  

• 10分钟作业直接通过SQL或DataFrame API读取Hive表：

  cached_df = spark.sql("SELECT * FROM your_db.cached_table_name")
  # 或
  cached_df = spark.read.table("your_db.cached_table_name")
  

• 优势：表结构可持久化管理，支持SQL查询，如果你用Azure Synapse，内置的Hive Metastore无需额外部署，直接可用。

3. 补充：添加数据库连接重试机制（减少临时失败影响）

即使做了缓存，也可以给原有的SQL读取逻辑加重试，应对偶尔的连接波动：

• 用重试库实现（比如Python的tenacity）：

  from tenacity import retry, stop_after_attempt, wait_exponential
  
  @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
  def read_sql_table(table_name):
      return spark.read.format("jdbc") \
          .option("url", "jdbc:sqlserver://your-server.database.windows.net:1433;database=your-db") \
          .option("dbtable", table_name) \
          .option("user", "your-user") \
          .option("password", "your-pwd") \
          .load()
  

• 也可以在ADF管道层面给Spark活动配置重试策略，进一步降低失败概率。

方案优先级推荐 #

优先选方案1（Azure存储+Parquet），因为它最简单、成本最低，完全满足你的每日缓存需求；如果需要更规范的表结构管理，再考虑方案2。

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