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

# 연결 생성 (":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()

# 쿼리 실행
cur.execute("SELECT number, toString(number) as str FROM system.numbers LIMIT 3")

# 다양한 방식으로 결과 조회
print(cur.fetchone())    # Single row: (0, '0')
print(cur.fetchmany(2))  # Multiple rows: ((1, '1'), (2, '2'))

# 메타데이터 조회
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")

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는 형식이 지정된 인수, 자동 유형 추론, 구성 가능한 NULL 및 예외 처리를 지원하는 네이티브 인프로세스 Python UDF를 제공합니다.

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 처리, 예외 처리, DateTime 지원에 관한 전체 가이드는 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 리포지토리