혼합된 JSON 객체를 포함한 DataFrame

요구 사항

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

# 연결 생성 (기본값: 인메모리)
conn = chdb.connect(":memory:")
# 파일 기반 사용 시: conn = chdb.connect("mydata.db")

# 쿼리 실행을 위한 커서 생성
cur = conn.cursor()

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

# 다양한 방식으로 결과 가져오기
print(cur.fetchone())    # 단일 행: (0, '0')
print(cur.fetchmany(2))  # 여러 행: ((1, '1'), (2, '2'))

# 메타데이터 조회
print(cur.column_names())  # ['number', 'str']
print(cur.column_types())  # ['UInt64', 'String']

# 커서를 이터레이터로 사용
for row in cur:
    print(row)

# 리소스는 반드시 닫기
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()  # 선택 사항 - 객체 삭제 시 자동으로 닫힙니다

고급 세션 기능

# 사용자 지정 설정을 사용한 세션
sess = session.Session(
    path="/tmp/analytics_db",
)

# 쿼리 성능 최적화
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
)

사용자 정의 함수 (UDF)

사용자 정의 Python 함수로 SQL을 확장합니다:

기본적인 UDF 사용법

from chdb.udf import chdb_udf
from chdb import query

# 간단한 수학 함수
@chdb_udf()
def add_numbers(a, b):
    return int(a) + int(b)

# 문자열 처리 함수
@chdb_udf()
def reverse_string(text):
    return text[::-1]

# JSON 처리 함수  
@chdb_udf()
def extract_json_field(json_str, field):
    import json
    try:
        data = json.loads(json_str)
        return str(data.get(field, ''))
    except:
        return ''

# 쿼리에서 UDF 사용
result = query("""
    SELECT 
        add_numbers('10', '20') as sum_result,
        reverse_string('hello') as reversed,
        extract_json_field('{"name": "John", "age": 30}', 'name') as name
""")
print(result)

사용자 지정 반환 타입을 갖는 고급 UDF

# 특정 반환 유형을 가진 UDF
@chdb_udf(return_type="Float64")
def calculate_bmi(height_str, weight_str):
    height = float(height_str) / 100  # cm를 미터로 변환
    weight = float(weight_str)
    return weight / (height * height)

# 데이터 유효성 검사를 위한 UDF
@chdb_udf(return_type="UInt8") 
def is_valid_email(email):
    import re
    pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
    return 1 if re.match(pattern, email) else 0

# 복잡한 쿼리에서 사용
result = query("""
    SELECT 
        name,
        calculate_bmi(height, weight) as bmi,
        is_valid_email(email) as has_valid_email
    FROM (
        SELECT 
            'John' as name, '180' as height, '75' as weight, 'john@example.com' as email
        UNION ALL
        SELECT 
            'Jane' as name, '165' as height, '60' as weight, 'invalid-email' as email
    )
""", "Pretty")
print(result)

UDF 모범 사례

상태를 유지하지 않는 함수: UDF는 부작용이 없는 순수 함수여야 합니다
함수 내부에서 import: 필요한 모든 모듈은 UDF 내부에서 import해야 합니다
String 입출력: 모든 UDF 매개변수는 문자열입니다(TabSeparated 포맷)
오류 처리: 안정적인 UDF를 위해 try-catch 블록을 포함하십시오
성능: UDF는 각 행마다 호출되므로 성능을 최적화해야 합니다

# 오류 처리가 포함된 잘 구조화된 UDF
@chdb_udf(return_type="String")
def safe_json_extract(json_str, path):
    import json
    try:
        data = json.loads(json_str)
        keys = path.split('.')
        result = data
        for key in keys:
            if isinstance(result, dict) and key in result:
                result = result[key]
            else:
                return 'null'
        return str(result)
    except Exception as e:
        return f'error: {str(e)}'

# 복잡한 중첩 JSON과 함께 사용
query("""
    SELECT safe_json_extract(
        '{"user": {"profile": {"name": "Alice", "age": 25}}}',
        'user.profile.name'
    ) as extracted_name
""")

스트리밍 쿼리 처리

메모리 사용량을 일정하게 유지하면서 대규모 데이터 집합을 처리합니다:

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):
    """데이터베이스 유형의 데이터 소스를 위한 커스텀 리더"""
    
    def __init__(self, connection_string: str):
        # 데이터베이스 연결 시뮬레이션
        self.data = self._load_data(connection_string)
        self.cursor = 0
        self.batch_size = 1000
        super().__init__(self.data)
    
    def _load_data(self, conn_str):
        # 데이터베이스에서 로딩 시뮬레이션
        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]]:
        """명시적 타입으로 테이블 스키마 정의"""
        return [
            ("id", "UInt64"),
            ("name", "String"),
            ("score", "Int64"),
            ("metadata", "String")  # JSON을 문자열로 저장
        ]
    
    def read(self, col_names: List[str], count: int) -> List[List[Any]]:
        """배치 단위로 데이터 읽기"""
        if self.cursor >= len(self.data["id"]):
            return []  # 데이터 없음
        
        end_pos = min(self.cursor + min(count, self.batch_size), len(self.data["id"]))
        
        # 요청된 컬럼의 데이터 반환
        result = []
        for col in col_names:
            if col in self.data:
                result.append(self.data[col][self.cursor:end_pos])
            else:
                # 누락된 컬럼 처리
                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

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"}]}
    ]
})

# 설정을 통한 JSON 타입 추론 제어
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  -- JSON 감지를 위해 전체 행 분석
""", "Pretty")
print(result)

# 고급 JSON 연산
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 리포지토리에 보고하십시오

​요구 사항

​설치

​사용 방법

​명령줄 인터페이스

​Python 기본 사용법

​연결 기반 API (권장)

​데이터 입력 방식

​파일 기반 데이터 소스

​출력 형식 예시

​DataFrame 연산

​기존 DataFrame API

​Python 테이블 엔진(권장)

​상태 유지 세션

​고급 세션 기능

​Python DB-API 2.0 인터페이스

​사용자 정의 함수 (UDF)

​기본적인 UDF 사용법

​사용자 지정 반환 타입을 갖는 고급 UDF

​UDF 모범 사례

​스트리밍 쿼리 처리

​Python 테이블 엔진

​Pandas DataFrame 쿼리하기

​PyReader를 사용한 사용자 지정 데이터 소스

​성능 및 최적화

​벤치마크

​성능 팁

​GitHub 리포지토리