DataStore 실행 모델

DataStore의 지연 평가, 실행 트리거, 캐싱 이해하기

DataStore를 효과적으로 활용하고 최적의 성능을 얻으려면 지연 평가 모델을 이해하는 것이 중요합니다.

​

지연 평가

DataStore는 지연 평가를 사용합니다. 즉, 연산은 즉시 실행되지 않고 기록된 후 최적화된 SQL 쿼리로 컴파일됩니다. 실제로 결과가 필요할 때만 실행됩니다.

​

예시: 지연 평가와 즉시 평가

from pathlib import Path
Path("sales.csv").write_text("""\
region,product,category,amount,quantity,price,date,order_id
East,Widget,Electronics,5200,10,120,2024-01-15,1001
West,Gadget,Electronics,800,5,160,2024-02-20,1002
East,Gizmo,Home,6500,3,100,2024-03-10,1003
North,Widget,Electronics,4500,6,150,2024-06-18,1004
West,Gadget,Electronics,2000,8,250,2024-09-14,1005
""")

from chdb import datastore as pd

ds = pd.read_csv("sales.csv")

# 이 연산들은 아직 실행되지 않습니다
result = (ds
    .filter(ds['amount'] > 1000)    # 기록됨, 실행되지 않음
    .select('region', 'amount')      # 기록됨, 실행되지 않음
    .groupby('region')               # 기록됨, 실행되지 않음
    .agg({'amount': 'sum'})          # 기록됨, 실행되지 않음
    .sort('sum', ascending=False)    # 기록됨, 실행되지 않음
)

# 아직 실행되지 않음 - 쿼리 계획 구성 중
print(result.to_sql())
# SELECT region, SUM(amount) AS sum
# FROM file('sales.csv', 'CSVWithNames')
# WHERE amount > 1000
# GROUP BY region
# ORDER BY sum DESC

# 여기서 실행됩니다
df = result.to_df()  # <-- 실행 트리거

​

지연 평가의 이점

1. 쿼리 최적화: 여러 작업이 단일한 최적화된 SQL 쿼리로 컴파일됩니다
2. 필터 푸시다운: 필터가 데이터 소스 수준에서 적용됩니다
3. 컬럼 프루닝: 필요한 컬럼만 읽습니다
4. 결정 지연: 실행 엔진을 런타임에 선택할 수 있습니다
5. 계획 검토: 실행 전에 쿼리를 확인하거나 디버깅할 수 있습니다

---

​

실행 트리거

실제 값이 필요할 때 실행이 자동으로 트리거됩니다:

​

자동 실행 트리거

Trigger	Example	Description
print() / repr()	print(ds)	결과 출력
len()	len(ds)	행 수 확인
.columns	ds.columns	컬럼 이름 확인
.dtypes	ds.dtypes	컬럼 타입 확인
.shape	ds.shape	크기 확인
.index	ds.index	행 인덱스 확인
.values	ds.values	NumPy 배열 가져오기
Iteration	for row in ds	행 순회
to_df()	ds.to_df()	pandas로 변환
to_pandas()	ds.to_pandas()	to_df의 별칭
to_dict()	ds.to_dict()	dict로 변환
to_numpy()	ds.to_numpy()	배열로 변환
.equals()	ds.equals(other)	DataStore 비교

예시:

# 아래 모든 코드는 실행을 트리거합니다
print(ds)              # 출력
len(ds)                # 1000
ds.columns             # Index(['name', 'age', 'city'])
ds.shape               # (1000, 3)
list(ds)               # 값 목록
ds.to_df()             # pandas DataFrame으로 변환

​

지연 상태를 유지하는 연산

Operation	Returns	Description
filter()	DataStore	WHERE 절 추가
select()	DataStore	컬럼 선택 추가
sort()	DataStore	ORDER BY 추가
groupby()	LazyGroupBy	GROUP BY 준비
join()	DataStore	JOIN 추가
ds['col']	ColumnExpr	컬럼 참조
ds[['col1', 'col2']]	DataStore	컬럼 선택

예시:

# 이 연산들은 실행을 트리거하지 않습니다 - 지연 상태를 유지합니다
result = ds.filter(ds['age'] > 25)      # DataStore 반환
result = ds.select('name', 'age')        # DataStore 반환
result = ds['name']                      # ColumnExpr 반환
result = ds.groupby('city')              # LazyGroupBy 반환

---

​

3단계 실행

DataStore 작업은 3단계 실행 모델을 따릅니다:

​

단계 1: SQL 쿼리 생성 (지연)

SQL로 표현할 수 있는 작업이 누적됩니다:

result = (ds
    .filter(ds['status'] == 'active')   # WHERE
    .select('user_id', 'amount')         # SELECT
    .groupby('user_id')                  # GROUP BY
    .agg({'amount': 'sum'})              # SUM()
    .sort('sum', ascending=False)        # ORDER BY
    .limit(10)                           # LIMIT
)
# 모두 하나의 SQL 쿼리로 컴파일됨

​

단계 2: 실행 시점

트리거가 발생하면 누적된 SQL이 실행됩니다:

# 여기서 실행이 트리거됩니다
df = result.to_df()  
# 최적화된 단일 SQL 쿼리가 이 시점에 실행됩니다

​

단계 3: DataFrame 작업 (해당하는 경우)

실행 후 pandas 전용 작업을 이어서 수행하면:

# 혼합 연산
result = (ds
    .filter(ds['amount'] > 100)          # 단계 1: SQL
    .to_df()                             # 단계 2: 실행
    .pivot_table(...)                    # 단계 3: pandas
)

---

​

실행 계획 보기

무엇이 실행되는지 확인하려면 explain()을 사용하세요:

Query

ds = pd.read_csv("sales.csv")

query = (ds
    .filter(ds['amount'] > 1000)
    .groupby('region')
    .agg({'amount': ['sum', 'mean']})
)

# 실행 계획 보기
query.explain()

Response

Pipeline:
  1. Source: file('sales.csv', 'CSVWithNames')
  2. Filter: amount > 1000
  3. GroupBy: region
  4. Aggregate: sum(amount), avg(amount)

Generated SQL:
SELECT region, SUM(amount) AS sum, AVG(amount) AS mean
FROM file('sales.csv', 'CSVWithNames')
WHERE amount > 1000
GROUP BY region

더 자세한 내용은 verbose=True를 사용하십시오:

query.explain(verbose=True)

디버깅: explain()에서 전체 문서를 참조하십시오.

---

​

캐싱

DataStore는 불필요한 쿼리를 방지하기 위해 실행 결과를 캐시합니다.

​

캐싱 동작 방식

from pathlib import Path
Path("data.csv").write_text("""\
name,age,city,salary,department
Alice,25,NYC,55000,Engineering
Bob,30,LA,65000,Product
Charlie,35,NYC,80000,Engineering
Diana,28,SF,70000,Design
Eve,42,NYC,95000,Product
""")

ds = pd.read_csv("data.csv")
result = ds.filter(ds['age'] > 25)

# 첫 번째 접근 - 쿼리 실행
print(result.shape)  # 실행 후 캐시에 저장

# 두 번째 접근 - 캐시 사용
print(result.columns)  # 캐시된 결과 사용

# 세 번째 접근 - 캐시 사용
df = result.to_df()  # 캐시된 결과 사용

​

캐시 무효화

DataStore를 변경하는 작업이 수행되면 캐시가 무효화됩니다:

result = ds.filter(ds['age'] > 25)
print(result.shape)  # 실행 후 캐시에 저장

# 새 연산으로 캐시 무효화
result2 = result.filter(result['city'] == 'NYC')
print(result2.shape)  # 재실행 (다른 쿼리)

​

수동 캐시 제어

# 캐시 지우기
ds.clear_cache()

# 캐싱 비활성화
from chdb.datastore.config import config
config.set_cache_enabled(False)

---

​

SQL과 Pandas 작업 함께 사용하기

DataStore는 SQL과 Pandas를 함께 사용하는 작업을 지능적으로 처리합니다:

​

SQL 호환 연산

다음은 SQL로 변환됩니다:

• filter(), where()
• select()
• groupby(), agg()
• sort(), orderby()
• limit(), offset()
• join(), union()
• distinct()
• 컬럼 연산(수학 연산, 비교, 문자열 메서드)

​

Pandas 전용 작업

다음 작업은 실행을 유발하며 pandas를 사용합니다:

• 사용자 지정 함수를 사용하는 apply()
• 복잡한 집계를 사용하는 pivot_table()
• stack(), unstack()
• 실행된 DataFrame에 대한 작업

​

하이브리드 파이프라인

# SQL 단계
result = (ds
    .filter(ds['amount'] > 100)      # SQL
    .groupby('category')              # SQL
    .agg({'amount': 'sum'})           # SQL
)

# 실행 + pandas 단계
result = (result
    .to_df()                          # SQL 실행
    .pivot_table(...)                 # pandas 연산
)

---

​

실행 엔진 선택

DataStore는 다양한 엔진을 사용해 작업을 실행할 수 있습니다:

​

자동 모드 (기본값)

from chdb.datastore.config import config

config.set_execution_engine('auto')  # 기본값
# 작업별로 최적의 엔진을 자동으로 선택합니다

​

chDB Engine 사용 강제

config.set_execution_engine('chdb')
# 모든 작업에 ClickHouse SQL 사용

​

pandas Engine 강제 설정

config.set_execution_engine('pandas')
# 모든 작업에 pandas 사용

자세한 내용은 구성: 실행 엔진을 참고하십시오.

---

​

성능에 미치는 영향

​

좋은 방법: 초기에 필터링

# 좋은 예: SQL에서 필터링 후 집계
result = (ds
    .filter(ds['date'] >= '2024-01-01')  # 초기에 데이터 범위를 줄임
    .groupby('category')
    .agg({'amount': 'sum'})
)

​

나쁨: 필터를 나중에 적용

# 나쁜 예: 전체 집계 후 필터링
result = (ds
    .groupby('category')
    .agg({'amount': 'sum'})
    .to_df()
    .query('sum > 1000')  # 집계 후 Pandas 필터링
)

​

권장: 초기에 컬럼 선택하기

# 좋은 예: SQL에서 컬럼 선택
result = (ds
    .select('user_id', 'amount', 'date')
    .filter(ds['date'] >= '2024-01-01')
    .groupby('user_id')
    .agg({'amount': 'sum'})
)

​

좋은 예: SQL이 작업을 처리하게 하세요

# 좋은 예: SQL에서 복잡한 집계 처리
result = (ds
    .groupby('category')
    .agg({
        'amount': ['sum', 'mean', 'count'],
        'quantity': 'sum'
    })
    .sort('sum', ascending=False)
    .limit(10)
)
# 하나의 SQL 쿼리로 모든 작업 처리

# 나쁜 예: 여러 개의 별도 쿼리
sums = ds.groupby('category')['amount'].sum().to_df()
means = ds.groupby('category')['amount'].mean().to_df()
# 하나 대신 두 개의 쿼리 실행

---

​

모범 사례 요약

1. 실행하기 전에 작업을 체이닝하세요 - 전체 쿼리를 구성한 뒤 한 번만 트리거하세요
2. 초기에 필터링하세요 - 소스에서 데이터를 줄이세요
3. 필요한 컬럼만 선택하세요 - 컬럼 프루닝은 성능을 향상시킵니다
4. 실행 과정을 이해하려면 explain()을 사용하세요 - 실행 전에 디버깅하세요
5. 집계는 SQL에 맡기세요 - ClickHouse는 이에 최적화되어 있습니다
6. 실행 트리거를 유의하세요 - 의도치 않게 일찍 실행되지 않도록 하세요
7. 캐싱을 현명하게 사용하세요 - 캐시가 언제 무효화되는지 이해하세요