为 Python 安装 chDB

要求

Python 3.8+
支持的平台：macOS 和 Linux (x86_64 和 ARM64)

安装

pip install chdb

使用

命令行界面

可直接在命令行中运行 SQL 查询：

# 基础查询
python3 -m chdb "SELECT 1, 'abc'" Pretty

# 带格式化的查询
python3 -m chdb "SELECT version()" JSON

Python 基础用法

import chdb

# 简单查询
result = chdb.query("SELECT 1 as id, 'Hello World' as message", "CSV")
print(result)

# 获取查询统计信息
print(f"Rows read: {result.rows_read()}")
print(f"Bytes read: {result.bytes_read()}")
print(f"Execution time: {result.elapsed()} seconds")

基于连接的 API (推荐)

为了更好地管理资源并提升性能：

import chdb

# Create connection (":memory:" is a temporary, non-persistent database)
conn = chdb.connect(":memory:")
# Or use file-based: conn = chdb.connect("mydata.db")

# Create cursor for query execution
cur = conn.cursor()

# Execute queries
cur.execute("SELECT number, toString(number) as str FROM system.numbers LIMIT 3")

# Fetch results in different ways
print(cur.fetchone())    # Single row: (0, '0')
print(cur.fetchmany(2))  # Multiple rows: ((1, '1'), (2, '2'))

# Get metadata
print(cur.column_names())  # ['number', 'str']
print(cur.column_types())  # ['UInt64', 'String']

# Use cursor as iterator
for row in cur:
    print(row)

# Always close resources
cur.close()
conn.close()

数据输入方式

基于文件的数据源

chDB 支持 70 多种数据格式，可直接对文件进行查询：

import chdb
# 准备您的数据
# ...

# 查询 Parquet 文件
result = chdb.query("""
    SELECT customer_id, sum(amount) as total
    FROM file('sales.parquet', Parquet) 
    GROUP BY customer_id 
    ORDER BY total DESC 
    LIMIT 10
""", 'JSONEachRow')

# 查询带表头的 CSV 文件
result = chdb.query("""
    SELECT * FROM file('data.csv', CSVWithNames) 
    WHERE column1 > 100
""", 'DataFrame')

# 多种文件格式
result = chdb.query("""
    SELECT * FROM file('logs*.jsonl', JSONEachRow)
    WHERE timestamp > '2024-01-01'
""", 'Pretty')

输出格式示例

# 用于分析的 DataFrame
df = chdb.query('SELECT * FROM system.numbers LIMIT 5', 'DataFrame')
print(type(df))  # <class 'pandas.core.frame.DataFrame'>

# 用于互操作性的 Arrow Table  
arrow_table = chdb.query('SELECT * FROM system.numbers LIMIT 5', 'ArrowTable')
print(type(arrow_table))  # <class 'pyarrow.lib.Table'>

# 用于 API 的 JSON
json_result = chdb.query('SELECT version()', 'JSON')
print(json_result)

# 用于调试的 Pretty 格式
pretty_result = chdb.query('SELECT * FROM system.numbers LIMIT 3', 'Pretty')
print(pretty_result)

DataFrame 操作

旧版 DataFrame API

import chdb.dataframe as cdf
import pandas as pd

# 连接多个 DataFrame
df1 = pd.DataFrame({'a': [1, 2, 3], 'b': ["one", "two", "three"]})
df2 = pd.DataFrame({'c': [1, 2, 3], 'd': ["①", "②", "③"]})

result_df = cdf.query(
    sql="SELECT * FROM __tbl1__ t1 JOIN __tbl2__ t2 ON t1.a = t2.c",
    tbl1=df1, 
    tbl2=df2
)
print(result_df)

# 查询结果 DataFrame
summary = result_df.query('SELECT b, sum(a) FROM __table__ GROUP BY b')
print(summary)

Python 表引擎 (推荐)

import chdb
import pandas as pd
import pyarrow as pa

# 直接查询 Pandas DataFrame
df = pd.DataFrame({
    "customer_id": [1, 2, 3, 1, 2],
    "product": ["A", "B", "A", "C", "A"],
    "amount": [100, 200, 150, 300, 250],
    "metadata": [
        {'category': 'electronics', 'priority': 'high'},
        {'category': 'books', 'priority': 'low'},
        {'category': 'electronics', 'priority': 'medium'},
        {'category': 'clothing', 'priority': 'high'},
        {'category': 'books', 'priority': 'low'}
    ]
})

# 支持 JSON 的 DataFrame 直接查询
result = chdb.query("""
    SELECT 
        customer_id,
        sum(amount) as total_spent,
        toString(metadata.category) as category
    FROM Python(df)
    WHERE toString(metadata.priority) = 'high'
    GROUP BY customer_id, toString(metadata.category)
    ORDER BY total_spent DESC
""").show()

# 查询 Arrow Table
arrow_table = pa.table({
    "id": [1, 2, 3, 4],
    "name": ["Alice", "Bob", "Charlie", "David"],
    "score": [98, 89, 86, 95]
})

chdb.query("""
    SELECT name, score
    FROM Python(arrow_table)
    ORDER BY score DESC
""").show()

有状态会话

会话可在多个操作之间保留查询状态，从而支持更复杂的工作流：

from chdb import session

# 临时会话（自动清理）
sess = session.Session()

# 或使用指定路径的持久化会话
# sess = session.Session("/path/to/data")

# 创建数据库和表
sess.query("CREATE DATABASE IF NOT EXISTS analytics ENGINE = Atomic")
sess.query("USE analytics")

sess.query("""
    CREATE TABLE sales (
        id UInt64,
        product String,
        amount Decimal(10,2),
        sale_date Date
    ) ENGINE = MergeTree() 
    ORDER BY (sale_date, id)
""")

# 插入数据
sess.query("""
    INSERT INTO sales VALUES 
        (1, 'Laptop', 999.99, '2024-01-15'),
        (2, 'Mouse', 29.99, '2024-01-16'),
        (3, 'Keyboard', 79.99, '2024-01-17')
""")

# 创建 materialized view
sess.query("""
    CREATE MATERIALIZED VIEW daily_sales AS
    SELECT 
        sale_date,
        count() as orders,
        sum(amount) as revenue
    FROM sales 
    GROUP BY sale_date
""")

# 查询视图
result = sess.query("SELECT * FROM daily_sales ORDER BY sale_date", "Pretty")
print(result)

# 会话自动管理资源
sess.close()  # 可选——对象销毁时自动关闭

高级会话功能

# Session with custom settings
sess = session.Session(
    path="/tmp/analytics_db",
)

# Query performance optimization
result = sess.query("""
    SELECT product, sum(amount) as total
    FROM sales 
    GROUP BY product
    ORDER BY total DESC
    SETTINGS max_threads = 4
""", "JSON")

另见：test_stateful.py。

Python DB-API 2.0 接口

与现有 Python 应用程序兼容的标准数据库接口：

import chdb.dbapi as dbapi

# 检查驱动程序信息
print(f"chDB driver version: {dbapi.get_client_info()}")

# 创建连接
conn = dbapi.connect()
cursor = conn.cursor()

# 使用参数执行查询
cursor.execute("""
    SELECT number, number * ? as doubled 
    FROM system.numbers 
    LIMIT ?
""", (2, 5))

# 获取元数据
print("Column descriptions:", cursor.description)
print("Row count:", cursor.rowcount)

# 获取结果
print("First row:", cursor.fetchone())
print("Next 2 rows:", cursor.fetchmany(2))

# 获取剩余行
for row in cursor.fetchall():
    print("Row:", row)

# 批次操作
data = [(1, 'Alice'), (2, 'Bob'), (3, 'Charlie')]
cursor.execute("""
    CREATE TABLE temp_users (
        id UInt64,
        name String
    ) ENGINE = MergeTree()
    ORDER BY (id)
""")
cursor.executemany(
    "INSERT INTO temp_users (id, name) VALUES (?, ?)", 
    data
)

Python 用户自定义函数 (UDF)

chDB 支持原生进程内 Python UDF，具有类型化参数、自动类型推断以及可配置的 NULL 和异常处理机制。

from chdb import query, func
from chdb.sqltypes import INT64

@func([INT64, INT64], INT64)
def add(a, b):
    return a + b

result = query("SELECT add(2, 3)")
print(result)  # 5

有关注册方法、类型系统、NULL 处理、异常处理和日期时间支持的完整说明，请参阅 Python UDF 指南。有关完整的 API 参考文档，请参阅 Python UDF API 参考文档。较早的 @chdb_udf 装饰器仍可使用，但已由此 API 替代——请参阅旧版 API。

流式查询处理

以恒定的内存占用处理大型数据集：

from chdb import session

sess = session.Session()

# 设置大型数据集
sess.query("""
    CREATE TABLE large_data ENGINE = Memory() AS 
    SELECT number as id, toString(number) as data 
    FROM numbers(1000000)
""")

# 示例 1：使用上下文管理器进行基本流式处理
total_rows = 0
with sess.send_query("SELECT * FROM large_data", "CSV") as stream:
    for chunk in stream:
        chunk_rows = len(chunk.data().split('\n')) - 1
        total_rows += chunk_rows
        print(f"Processed chunk: {chunk_rows} rows")
        
        # 如有需要，提前终止
        if total_rows > 100000:
            break

print(f"Total rows processed: {total_rows}")

# 示例 2：手动迭代并显式释放资源
stream = sess.send_query("SELECT * FROM large_data WHERE id % 100 = 0", "JSONEachRow")
processed_count = 0

while True:
    chunk = stream.fetch()
    if chunk is None:
        break
    
    # 处理数据块
    lines = chunk.data().strip().split('\n')
    for line in lines:
        if line:  # 跳过空行
            processed_count += 1
    
    print(f"Processed {processed_count} records so far...")
    
stream.close()  # 重要：显式释放资源

# 示例 3：与外部库集成（Arrow 格式）
import pyarrow as pa
from deltalake import write_deltalake

# 以 Arrow 格式流式传输结果
stream = sess.send_query("SELECT * FROM large_data LIMIT 100000", "Arrow")

# 创建指定批次大小的 RecordBatchReader
batch_reader = stream.record_batch(rows_per_batch=10000)

# 导出到 Delta Lake
write_deltalake(
    table_or_uri="./my_delta_table",
    data=batch_reader,
    mode="overwrite"
)

stream.close()
sess.close()

Python 表引擎

查询 Pandas DataFrame 数据

import chdb
import pandas as pd

# 包含嵌套数据的复杂 DataFrame
df = pd.DataFrame({
    "customer_id": [1, 2, 3, 4, 5, 6],
    "customer_name": ["Alice", "Bob", "Charlie", "Alice", "Bob", "David"],
    "orders": [
        {"order_id": 101, "amount": 250.50, "items": ["laptop", "mouse"]},
        {"order_id": 102, "amount": 89.99, "items": ["book"]},
        {"order_id": 103, "amount": 1299.99, "items": ["phone", "case", "charger"]},
        {"order_id": 104, "amount": 45.50, "items": ["pen", "paper"]},
        {"order_id": 105, "amount": 199.99, "items": ["headphones"]},
        {"order_id": 106, "amount": 15.99, "items": ["cable"]}
    ]
})

# 使用 JSON 操作进行高级查询
result = chdb.query("""
    SELECT 
        customer_name,
        count() as order_count,
        sum(toFloat64(orders.amount)) as total_spent,
        arrayStringConcat(
            arrayDistinct(
                arrayFlatten(
                    groupArray(orders.items)
                )
            ), 
            ', '
        ) as all_items
    FROM Python(df)
    GROUP BY customer_name
    HAVING total_spent > 100
    ORDER BY total_spent DESC
""").show()

# 在 DataFrame 上使用窗口函数
window_result = chdb.query("""
    SELECT 
        customer_name,
        toFloat64(orders.amount) as amount,
        sum(toFloat64(orders.amount)) OVER (
            PARTITION BY customer_name 
            ORDER BY toInt32(orders.order_id)
        ) as running_total
    FROM Python(df)
    ORDER BY customer_name, toInt32(orders.order_id)
""", "Pretty")
print(window_result)

使用 PyReader 自定义数据源

为特定数据源实现自定义数据读取器：

import chdb
from typing import List, Tuple, Any
import json

class DatabaseReader(chdb.PyReader):
    """Custom reader for database-like data sources"""
    
    def __init__(self, connection_string: str):
        # Simulate database connection
        self.data = self._load_data(connection_string)
        self.cursor = 0
        self.batch_size = 1000
        super().__init__(self.data)
    
    def _load_data(self, conn_str):
        # Simulate loading from database
        return {
            "id": list(range(1, 10001)),
            "name": [f"user_{i}" for i in range(1, 10001)],
            "score": [i * 10 + (i % 7) for i in range(1, 10001)],
            "metadata": [
                json.dumps({"level": i % 5, "active": i % 3 == 0})
                for i in range(1, 10001)
            ]
        }
    
    def get_schema(self) -> List[Tuple[str, str]]:
        """Define table schema with explicit types"""
        return [
            ("id", "UInt64"),
            ("name", "String"),
            ("score", "Int64"),
            ("metadata", "String")  # JSON stored as string
        ]
    
    def read(self, col_names: List[str], count: int) -> List[List[Any]]:
        """Read data in batches"""
        if self.cursor >= len(self.data["id"]):
            return []  # No more data
        
        end_pos = min(self.cursor + min(count, self.batch_size), len(self.data["id"]))
        
        # Return data for requested columns
        result = []
        for col in col_names:
            if col in self.data:
                result.append(self.data[col][self.cursor:end_pos])
            else:
                # Handle missing columns
                result.append([None] * (end_pos - self.cursor))
        
        self.cursor = end_pos
        return result

### JSON Type Inference and Handling                                

chDB automatically handles complex nested data structures:

```python
import pandas as pd
import chdb

# DataFrame with mixed JSON objects
df_with_json = pd.DataFrame({
    "user_id": [1, 2, 3, 4],
    "profile": [
        {"name": "Alice", "age": 25, "preferences": ["music", "travel"]},
        {"name": "Bob", "age": 30, "location": {"city": "NYC", "country": "US"}},
        {"name": "Charlie", "skills": ["python", "sql", "ml"], "experience": 5},
        {"score": 95, "rank": "gold", "achievements": [{"title": "Expert", "date": "2024-01-01"}]}
    ]
})

# Control JSON inference with settings
result = chdb.query("""
    SELECT 
        user_id,
        profile.name as name,
        profile.age as age,
        length(profile.preferences) as pref_count,
        profile.location.city as city
    FROM Python(df_with_json)
    SETTINGS pandas_analyze_sample = 1000  -- Analyze all rows for JSON detection
""", "Pretty")
print(result)

# Advanced JSON operations
complex_json = chdb.query("""
    SELECT 
        user_id,
        JSONLength(toString(profile)) as json_fields,
        JSONType(toString(profile), 'preferences') as pref_type,
        if(
            JSONHas(toString(profile), 'achievements'),
            JSONExtractString(toString(profile), 'achievements[0].title'),
            'None'
        ) as first_achievement
    FROM Python(df_with_json)
""", "JSONEachRow")
print(complex_json)

性能与优化

基准测试

chDB 持续优于其他嵌入式引擎：

DataFrame 操作：在分析查询中，速度比传统 DataFrame 库快 2–5 倍
Parquet 处理：可与领先的列式引擎媲美
内存效率：内存占用低于替代方案

更多基准测试结果详情

性能建议

import chdb

# 1. 使用合适的输出格式
df_result = chdb.query("SELECT * FROM large_table", "DataFrame")  # 用于分析
arrow_result = chdb.query("SELECT * FROM large_table", "Arrow")    # 用于互操作
native_result = chdb.query("SELECT * FROM large_table", "Native")   # 用于 chDB 之间传输

# 2. 通过配置项优化查询
fast_result = chdb.query("""
    SELECT customer_id, sum(amount) 
    FROM sales 
    GROUP BY customer_id
    SETTINGS 
        max_threads = 8,
        max_memory_usage = '4G',
        use_uncompressed_cache = 1
""", "DataFrame")

# 3. 对大型数据集使用流式处理
from chdb import session

sess = session.Session()

# 初始化大型数据集
sess.query("""
    CREATE TABLE large_sales ENGINE = Memory() AS 
    SELECT 
        number as sale_id,
        number % 1000 as customer_id,
        rand() % 1000 as amount
    FROM numbers(10000000)
""")

# 以恒定内存占用进行流式处理
total_amount = 0
processed_rows = 0

with sess.send_query("SELECT customer_id, sum(amount) as total FROM large_sales GROUP BY customer_id", "JSONEachRow") as stream:
    for chunk in stream:
        lines = chunk.data().strip().split('\n')
        for line in lines:
            if line:  # 跳过空行
                import json
                row = json.loads(line)
                total_amount += row['total']
                processed_rows += 1
        
        print(f"Processed {processed_rows} customer records, running total: {total_amount}")
        
        # 演示用提前终止
        if processed_rows > 1000:
            break

print(f"Final result: {processed_rows} customers processed, total amount: {total_amount}")

# 流式传输至外部系统（例如 Delta Lake）
stream = sess.send_query("SELECT * FROM large_sales LIMIT 1000000", "Arrow")
batch_reader = stream.record_batch(rows_per_batch=50000)

# 按批次处理
for batch in batch_reader:
    print(f"Processing batch with {batch.num_rows} rows...")
    # 转换或导出每个批次
    # df_batch = batch.to_pandas()
    # process_batch(df_batch)

stream.close()
sess.close()

GitHub 仓库

主仓库：chdb-io/chdb
问题与支持：请在 GitHub 仓库中报告问题

入门

语言集成

DataStore

配置

调试

开发者指南

技术参考文档

要求

安装

使用

命令行界面

Python 基础用法

基于连接的 API (推荐)

数据输入方式

基于文件的数据源

输出格式示例

DataFrame 操作

旧版 DataFrame API

Python 表引擎 (推荐)

有状态会话

高级会话功能

Python DB-API 2.0 接口

Python 用户自定义函数 (UDF)

流式查询处理

Python 表引擎

查询 Pandas DataFrame 数据

使用 PyReader 自定义数据源

性能与优化

基准测试

性能建议

GitHub 仓库

​要求

​安装

​使用

​命令行界面

​Python 基础用法

​基于连接的 API (推荐)

​数据输入方式

​基于文件的数据源

​输出格式示例

​DataFrame 操作

​旧版 DataFrame API

​Python 表引擎 (推荐)

​有状态会话

​高级会话功能

​Python DB-API 2.0 接口

​Python 用户自定义函数 (UDF)

​流式查询处理

​Python 表引擎

​查询 Pandas DataFrame 数据

​使用 PyReader 自定义数据源

​性能与优化

​基准测试

​性能建议

​GitHub 仓库

要求

安装

使用

命令行界面

Python 基础用法

基于连接的 API (推荐)

数据输入方式

基于文件的数据源

输出格式示例

DataFrame 操作

旧版 DataFrame API

Python 表引擎 (推荐)

有状态会话

高级会话功能

Python DB-API 2.0 接口

Python 用户自定义函数 (UDF)

流式查询处理

Python 表引擎

查询 Pandas DataFrame 数据

使用 PyReader 自定义数据源

性能与优化

基准测试

性能建议

GitHub 仓库