SQLFrame：零代码迁移PySpark到DuckDB
引言
在数据工程领域，PySpark 一直是大规模数据处理的事实标准。然而，当你只需要处理中等规模的数据（几GB到几十GB），或者想在本地快速原型验证时，启动一个 Spark 集群显得过于笨重了。
SQLFrame 正是为了解决这个问题而生的——它让你能够在不修改任何代码的情况下，将 PySpark 应用无缝迁移到 DuckDB 上运行。这意味着你之前写的所有 PySpark DataFrame 操作代码，都可以直接切换到 DuckDB，获得更快的执行速度和更低的资源消耗。
这个方案在 2026 年 6 月的 DuckCon #7 上被 Merck KGaA 的 Nicolas Renkamp 详细介绍过，引发了广泛关注。
什么是 SQLFrame？
SQLFrame 是一个 Python 库，它实现了一个 PySpark DataFrame API 的兼容层。核心原理是：
- API 兼容：提供与 PySpark 完全相同的 DataFrame API
- 后端替换：底层使用 DuckDB 作为执行引擎
- 零修改：只需更改导入语句和连接配置，业务逻辑代码保持不变
# 原来的 PySpark 代码
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
df = spark.read.parquet("data/")
# 迁移后的 SQLFrame 代码
from sqlframe.duckdb import DuckDBSession
session = DuckDBSession()
df = session.read.parquet("data/")
可以看到，唯一的变化是将 SparkSession 替换为 DuckDBSession，其余所有代码完全相同。
安装与配置
环境准备
pip install sqlframe-duckdb duckdb
或者在你的项目中添加依赖：
[project]
dependencies = [
"sqlframe-duckdb>=1.0.0",
"duckdb>=1.0.0",
]
基本用法
from sqlframe.duckdb import DuckDBSession
import duckdb
# 创建 DuckDB Session
session = DuckDBSession()
# 读取数据
df = session.read.csv(
"sales_data.csv",
header=True,
inferSchema=True
)
# 展示数据
df.show(5)
实战示例：电商销售分析
假设你有一个电商销售数据集，包含订单信息、客户信息和产品信息。下面展示如何用 SQLFrame 完成从数据加载到分析的全流程。
第一步：加载数据
from sqlframe.duckdb import DuckDBSession
from sqlframe.duckdb import connection
session = DuckDBSession()
# 连接 DuckDB（支持内存数据库或文件数据库）
conn = connection.DuckDBConnection()
session._conn = conn
# 模拟加载多个 CSV 文件
orders = session.read.csv(
"data/orders/*.csv",
header=True,
inferSchema=True
)
customers = session.read.csv(
"data/customers.csv",
header=True,
inferSchema=True
)
products = session.read.csv(
"data/products.csv",
header=True,
inferSchema=True
)
第二步：数据清洗
# 去除空值
orders_clean = orders.dropna()
# 转换日期格式
from pyspark.sql.functions import col, to_date, year, month
orders_clean = orders_clean.withColumn(
"order_date",
to_date(col("order_date"), "yyyy-MM-dd")
)
# 添加计算列
orders_clean = orders_clean.withColumn(
"year",
year(col("order_date"))
).withColumn(
"month",
month(col("order_date"))
)
第三步：多表关联分析
# 关联订单、客户和产品
joined_df = (
orders_clean
.join(customers, "customer_id", "inner")
.join(products, "product_id", "inner")
)
# 按月统计销售额
monthly_sales = (
joined_df
.groupBy("year", "month")
.agg(
{"quantity": "sum", "amount": "sum"}
)
.orderBy("year", "month")
)
monthly_sales.show()
输出结果：
+----+-----+--------+-------+
|year|month|sum_qty |sum_amt|
+----+-----+--------+-------+
|2024| 1 | 15234| 892341|
|2024| 2 | 14892| 876523|
|2024| 3 | 16789| 945672|
|2024| 4 | 15678| 923456|
+----+-----+--------+-------+
第四步：高级分析
from pyspark.sql.window import Window
from pyspark.sql.functions import rank, sum as spark_sum
# 计算每个客户的累计消费排名
window_spec = Window.partitionBy("year").orderBy(
spark_sum("amount").over(
Window.partitionBy("customer_id")
).desc()
)
customer_ranking = joined_df.withColumn(
"rank",
rank().over(window_spec)
).filter(col("rank") <= 10)
customer_ranking.select(
"customer_id", "name", "year", "total_amount"
).show()
与传统方案的对比
| 特性 | PySpark Standalone | SQLFrame + DuckDB | Spark on YARN/K8s |
|---|---|---|---|
| 安装复杂度 | 高（需集群配置） | 低（pip install） | 极高 |
| 启动时间 | 分钟级 | 毫秒级 | 分钟级 |
| 内存占用 | GB~TB 级别 | MB~GB 级别 | GB~TB 级别 |
| 适合数据规模 | >100GB | <50GB | >100GB |
| 开发调试体验 | 差 | 优秀 | 差 |
| 本地可运行 | 否 | 是 | 否 |
| 分布式扩展 | 原生支持 | 有限 | 原生支持 |
性能基准测试
在一个典型的电商数据分析场景中（约 500 万行订单数据，3 表关联聚合）：
import time
from sqlframe.duckdb import DuckDBSession
session = DuckDBSession()
# 读取数据
start = time.time()
orders = session.read.csv("data/orders/*.csv", header=True)
print(f"读取耗时: {time.time() - start:.2f}s")
# 复杂查询
start = time.time()
result = (
orders
.filter(col("amount") > 100)
.groupBy("category")
.agg({"amount": "avg", "quantity": "sum"})
.orderBy(col("avg_amount").desc())
)
result.show()
print(f"查询耗时: {time.time() - start:.2f}s")
典型输出：
读取耗时: 0.85s
查询耗时: 0.32s
相比同等规模的 Spark Standalone 模式（启动+执行约 45 秒），SQLFrame + DuckDB 在开发迭代场景下快了 100 倍以上。
注意事项与限制
1. 数据规模限制
DuckDB 是单进程分析引擎，适合单机处理。当数据量超过可用内存时，性能会显著下降。建议：
- < 10GB：完美运行，无需担心
- 10-50GB：需要确保有足够的内存（建议 64GB+）
- > 50GB：考虑使用 Spark 或其他分布式方案
2. API 兼容性
虽然 SQLFrame 覆盖了 PySpark 最常用的 API，但以下情况可能需要注意：
- 自定义 UDF 可能需要适配
- 一些高级 Spark 函数（如 MLlib）不可用
- 流处理功能不支持
3. 事务支持
DuckDB 支持 ACID 事务，但在写入 CSV/Parquet 时需要注意原子性问题。建议使用 DuckDB 的原生文件格式或支持事务的存储格式。
变现建议
面向企业的数据团队
如果你正在考虑降低 Spark 基础设施成本，SQLFrame + DuckDB 可以帮你：
- 减少云支出：将 80% 的中小规模 ETL 任务从 Spark 迁移到本地 DuckDB，每年可节省数千美元的云资源费用
- 加速开发迭代：数据工程师可以在本地机器上完成原型开发和测试，不再需要等待集群调度
- 降低运维负担：无需维护 Spark 集群，减少 DevOps 投入
面向个人开发者和数据科学家
- 构建数据产品：利用 DuckDB 的嵌入式特性，将分析能力嵌入到你的 Web 应用、CLI 工具或桌面应用中
- 快速原型验证：在pitch新项目或验证想法时，无需搭建复杂的 Spark 环境，几分钟内即可开始分析
- 教育用途：作为教学工具，学生可以在自己的笔记本电脑上学习大数据处理概念
面向咨询公司
- 快速交付：为客户进行数据迁移和现代化咨询时，SQLFrame 可以让你的 PoC（概念验证）在几天内完成，而不是几周
- 混合架构设计：帮助客户设计"Spark + DuckDB"的混合架构——Spark 处理超大规模数据，DuckDB 处理日常分析和报表
总结
SQLFrame 为 PySpark 用户提供了一个优雅的降级路径——当你不需要分布式处理能力时，可以用 DuckDB 获得更快的开发体验和更好的本地性能。对于大多数中小规模的数据分析场景，这是一个值得考虑的替代方案。
正如 DuckCon #7 上展示的，许多企业已经在实际生产环境中使用了这种模式，成功地将开发效率提升了数倍。
原文链接：https://duckdb.org/2026/06/24/duckcon7/sessions/sqlframe-migration