从 pandas 迁移 - ClickHouse Documentation

本指南可帮助您将现有 pandas 代码迁移到 DataStore，在保持兼容性的同时提升性能。

一行迁移

最简单的迁移方式是只需修改导入语句：

# 之前 (pandas)
import pandas as pd

# 之后 (DataStore)
from chdb import datastore as pd

就是这样！大多数 pandas 代码都可以直接运行，无需修改。

逐步迁移

安装 chDB

pip install "chdb>=4.0"

修改导入语句

# 将这一行：
import pandas as pd

# 改为：
from chdb import datastore as pd

测试代码

运行现有代码。大多数操作都无需修改：

from chdb import datastore as pd

# 这些用法都保持不变
df = pd.read_csv("data.csv")
result = df[df['age'] > 25]
grouped = df.groupby('city')['salary'].mean()
df.to_csv("output.csv")

处理差异

少数操作的行为有所不同。请参阅下方的关键区别。

哪些内容无需改动

数据加载

# 以下方法均可正常使用
df = pd.read_csv("data.csv")
df = pd.read_parquet("data.parquet")
df = pd.read_json("data.json")
df = pd.read_excel("data.xlsx")

筛选

# 布尔索引
df[df['age'] > 25]
df[(df['age'] > 25) & (df['city'] == 'NYC')]

# query() 方法
df.query('age > 25 and salary > 50000')

选区

# 列选择
df['name']
df[['name', 'age']]

# 行选择
df.head(10)
df.tail(10)
df.iloc[0:100]

GroupBy 和聚合

# 分组聚合
df.groupby('city')['salary'].mean()
df.groupby(['city', 'dept']).agg({'salary': ['sum', 'mean']})

排序

df.sort_values('salary', ascending=False)
df.sort_values(['city', 'age'])

字符串操作

df['name'].str.upper()
df['name'].str.contains('John')
df['name'].str.len()

DateTime 操作

df['date'].dt.year
df['date'].dt.month
df['date'].dt.dayofweek

I/O 操作

df.to_csv("output.csv")
df.to_parquet("output.parquet")
df.to_json("output.json")

关键区别

1. 惰性求值

DataStore 的操作采用惰性求值——只有在需要结果时才会执行。 pandas:

# 立即执行
result = df[df['age'] > 25]
print(type(result))  # pandas.DataFrame

DataStore:

# 构建查询，尚未执行
result = ds[ds['age'] > 25]
print(type(result))  # DataStore（惰性）

# 需要数据时才执行
print(result)        # 触发执行
df = result.to_df()  # 触发执行

2. 返回类型

操作	pandas 返回值	DataStore 返回值
`df['col']`	Series	ColumnExpr (惰性)
`df[['a', 'b']]`	DataFrame	DataStore (惰性)
`df[condition]`	DataFrame	DataStore (惰性)
`df.groupby('x')`	GroupBy	LazyGroupBy

3. 没有 inplace 参数

DataStore 不支持 inplace=True。请始终使用返回值： pandas:

df.drop(columns=['col'], inplace=True)

DataStore：

ds = ds.drop(columns=['col'])  # 将结果赋值给变量

4. 比较 DataStore

pandas 无法识别 DataStore 对象，因此请使用 to_pandas() 来进行比较：

# 这可能无法按预期工作
df == ds  # pandas 无法识别 DataStore

# 请改用以下方式
df.equals(ds.to_pandas())

5. 行顺序

对于文件源 (如 SQL 数据库) ，DataStore 可能不会保留行顺序。请使用显式排序：

# pandas 保留顺序
df = pd.read_csv("data.csv")

# DataStore - 使用 sort 确保顺序
ds = pd.read_csv("data.csv")
ds = ds.sort('id')  # 显式排序

迁移方案

模式 1：读取-分析-写入

# pandas
import pandas as pd
df = pd.read_csv("data.csv")
result = df[df['amount'] > 100].groupby('category')['amount'].sum()
result.to_csv("output.csv")

# DataStore - 同样的代码同样适用！
from chdb import datastore as pd
df = pd.read_csv("data.csv")
result = df[df['amount'] > 100].groupby('category')['amount'].sum()
result.to_csv("output.csv")

模式 2：使用 pandas 操作 DataFrame

如果你需要 pandas 特有功能，请在最后再进行转换：

from chdb import datastore as pd

# 快速 DataStore 操作
ds = pd.read_csv("large_data.csv")
ds = ds.filter(ds['date'] >= '2024-01-01')
ds = ds.filter(ds['amount'] > 100)

# 转换为 pandas 以使用特定功能
df = ds.to_df()
df_pivoted = df.pivot_table(...)  # pandas 专用

模式 3：混合工作流

from chdb import datastore as pd
import pandas

# 从 DataStore 开始进行快速过滤
ds = pd.read_csv("huge_file.csv")  # 1000 万行
ds = ds.filter(ds['year'] == 2024)  # 快速 SQL 过滤
ds = ds.select('col1', 'col2', 'col3')  # 列裁剪

# 转换以执行 pandas 特有操作
df = ds.to_df()  # 现在仅约 10 万行
result = df.apply(complex_custom_function)  # pandas 操作

性能对比

对于大型数据集，DataStore 的速度明显更快：

操作	pandas	DataStore	加速比
GroupBy 计数	347ms	17ms	19.93x
复杂管道	2,047ms	380ms	5.39x
Filter+Sort+Head	1,537ms	350ms	4.40x
GroupBy 聚合	406ms	141ms	2.88x

基于 1000 万行数据的基准测试

迁移故障排查

问题：操作无法正常运行

某些 pandas 操作可能暂不受支持。请检查：

该操作是否在兼容性列表中？
尝试先转换为 pandas：ds.to_df().operation()

问题：结果不一致

启用调试日志，以了解具体发生了什么：

from chdb.datastore.config import config
config.enable_debug()

# 查看生成的 SQL
ds.filter(ds['x'] > 10).explain()

问题：性能较慢

检查你的执行方式：

# 不推荐：多次小批量执行
for i in range(1000):
    result = ds.filter(ds['id'] == i).to_df()

# 推荐：单次执行
result = ds.filter(ds['id'].isin(ids)).to_df()

问题：类型不匹配

DataStore 推断出的类型可能会有所不同：

# 检查类型
print(ds.dtypes)

# 强制转换
ds['col'] = ds['col'].astype('int64')

渐进式迁移策略

第 1 周：兼容性测试

# 保留两个导入
import pandas as pd
from chdb import datastore as ds

# 比较结果
pdf = pd.read_csv("data.csv")
dsf = ds.read_csv("data.csv")

# 验证两者是否一致
assert pdf.equals(dsf.to_pandas())

第 2 周：迁移简单脚本

先从这类脚本开始：

读取大文件
执行过滤和聚合
不使用自定义 apply 函数

第 3 周：应对复杂场景

对于包含自定义函数的脚本：

from chdb import datastore as pd

# 让 DataStore 处理繁重的工作
ds = pd.read_csv("data.csv")
ds = ds.filter(ds['year'] == 2024)  # SQL

# 转换以进行自定义处理
df = ds.to_df()
result = df.apply(my_custom_function)

第 4 周：完整迁移

将所有脚本切换为使用 DataStore 导入。

常见问题

我可以同时使用 pandas 和 DataStore 吗？

可以！你可以自由地在两者之间转换：

from chdb import datastore as ds
import pandas as pd

# DataStore 转 pandas
df = ds_result.to_pandas()

# pandas 转 DataStore  
ds = ds.DataFrame(pd_result)

我的测试还能通过吗？

大多数测试应该都能通过。对于比较类测试，请先转换为 pandas：

def test_my_function():
    result = my_function()
    expected = pd.DataFrame(...)
    pd.testing.assert_frame_equal(result.to_pandas(), expected)

我可以在 Jupyter 笔记本中使用 DataStore 吗？

可以！DataStore 可在 Jupyter 笔记本中使用：

from chdb import datastore as pd

ds = pd.read_csv("data.csv")
ds.head()  # 在 Jupyter 中显示效果良好

如何报告问题？

如果发现兼容性问题，请前往以下地址提交： https://github.com/chdb-io/chdb/issues

​一行迁移

​逐步迁移

​安装 chDB

​修改导入语句

​测试代码

​处理差异

​哪些内容无需改动

​数据加载

​筛选

​选区

​GroupBy 和聚合

​排序

​字符串操作

​DateTime 操作

​I/O 操作

​关键区别

​1. 惰性求值

​2. 返回类型

​3. 没有 inplace 参数

​4. 比较 DataStore

​5. 行顺序

​迁移方案

​模式 1：读取-分析-写入

​模式 2：使用 pandas 操作 DataFrame

​模式 3：混合工作流

​性能对比

​迁移故障排查

​问题：操作无法正常运行

​问题：结果不一致

​问题：性能较慢

​问题：类型不匹配

​渐进式迁移策略

​第 1 周：兼容性测试

​第 2 周：迁移简单脚本

​第 3 周：应对复杂场景

​第 4 周：完整迁移

​常见问题

​我可以同时使用 pandas 和 DataStore 吗？

​我的测试还能通过吗？

​我可以在 Jupyter 笔记本中使用 DataStore 吗？

​如何报告问题？