mcpField
C
chdb-sql

chdb-sql

github.com/clickhouse数据与数据库5629.1k 次安装1 周前更新
代码仓库

关于 chdb-sql

用于在 Python 中通过 chdb 执行 ClickHouse SQL 的技能，无需启动服务器。它为 Python 用户介绍单次查询、使用 Session 进行有状态分析，以及通过 DB-API 2.0 连接文件、数据库和云存储。

主要功能

  1. 01在 Python 中运行 ClickHouse SQL
  2. 02查询文件、数据库和云存储
  3. 03通过 Session 进行持久或内存分析
  4. 04使用 DB-API 2.0 连接
  5. 05跨数据源关联数据

适用场景

  1. 01在 Python 脚本中查询 Parquet 文件
  2. 02使用 Session 构建多步分析
  3. 03通过 DB-API 连接访问数据库

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/clickhouse/agent-skills --skill chdb-sql

安装后在 Claude Code 中输入/chdb-sql确认已加载。

包含的文件

除 SKILL.md 外，安装时一起下载的参考资料和脚本。

metadata.json
README.md
examples/
examples.md
references/
api-reference.md
sql-functions.md
table-functions.md
scripts/
verify_install.py

详细介绍

chdb SQL — 在 Python 进程中运行 ClickHouse

直接在 Python 中运行 ClickHouse SQL — 无需服务器。使用完整的 ClickHouse SQL 功能查询本地文件、远程数据库和云存储。

pip install chdb

决策树：选择合适的 API

1. One-off query on files or databases → chdb.query()
2. Multi-step analysis with tables      → Session
3. DB-API 2.0 connection                → chdb.connect()
4. Pandas-style DataFrame operations    → Use chdb-datastore skill instead

chdb.query() — 一行代码，查询任意数据

import chdb

chdb.query("SELECT * FROM file('data.parquet', Parquet) WHERE price > 100 LIMIT 10")       # local files
chdb.query("SELECT * FROM mysql('db:3306', 'shop', 'orders', 'root', 'pass')")              # databases
chdb.query("SELECT * FROM s3('s3://bucket/data.parquet', NOSIGN) LIMIT 10")                 # cloud storage
chdb.query("SELECT * FROM deltaLake('s3://bucket/delta/table', NOSIGN) LIMIT 10")           # data lakes

# Cross-source join
chdb.query("""
    SELECT u.name, o.amount FROM mysql('db:3306', 'crm', 'users', 'root', 'pass') AS u
    JOIN file('orders.parquet', Parquet) AS o ON u.id = o.user_id ORDER BY o.amount DESC
""")

data = {"name": ["Alice", "Bob"], "score": [95, 87]}
chdb.query("SELECT * FROM Python(data) ORDER BY score DESC")                                # Python data
df = chdb.query("SELECT * FROM numbers(10)", "DataFrame")                                   # output formats
chdb.query("SELECT toDate({d:String}) + number FROM numbers({n:UInt64})",
    "DataFrame", params={"d": "2025-01-01", "n": 30})                                      # parametrized

表函数 → table-functions.md | SQL 函数 → sql-functions.md | 完整 API → api-reference.md

Session — 有状态分析流水线

from chdb import session as chs
sess = chs.Session("./analytics_db")   # persistent; Session() for in-memory

sess.query("CREATE TABLE users ENGINE=MergeTree() ORDER BY id AS SELECT * FROM mysql('db:3306','crm','users','root','pass')")
sess.query("CREATE TABLE events ENGINE=MergeTree() ORDER BY (ts,user_id) AS SELECT * FROM s3('s3://logs/events/*.parquet',NOSIGN)")
sess.query("""
    SELECT u.country, count() AS cnt, uniqExact(e.user_id) AS users
    FROM events e JOIN users u ON e.user_id = u.id
    WHERE e.ts >= today() - 7 GROUP BY u.country ORDER BY cnt DESC
""", "Pretty").show()
sess.close()

Connection API（DB-API 2.0）

from chdb import dbapi
conn = dbapi.connect()
cur = conn.cursor()
cur.execute("SELECT * FROM file('data.parquet', Parquet) WHERE value > 100")
print(cur.fetchall())
cur.close()
conn.close()

故障排查

问题解决方法
ImportError: No module named 'chdb'pip install chdb
DB::Exception: FILE_NOT_FOUND检查文件路径；使用绝对路径或确认当前工作目录
DB::Exception: Unknown table function检查函数名称拼写（例如，deltaLake 而不是 deltalake）
连接远程数据库时被拒绝检查 host:port 格式；确保远程数据库允许连接
环境检查运行 python scripts/verify_install.py（在技能目录中）

参考资料

注意：本技能介绍如何使用 chdb SQL。 如需执行 pandas 风格的操作，请使用 chdb-datastore 技能。 如需为 chdb 源代码做贡献，请参阅项目根目录中的 CLAUDE.md。

评论

支持 Markdown