Модель выполнения 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. Pushdown фильтров: Фильтры применяются на уровне источника данных
3. Исключение лишних столбцов: Считываются только нужные столбцы
4. Отложенный выбор: Движок выполнения можно выбрать во время выполнения
5. Просмотр плана: Перед выполнением можно просмотреть и отладить запрос

---

​

Триггеры выполнения

Выполнение автоматически запускается, когда требуются фактические значения:

​

Автоматические триггеры

Trigger	Пример	Описание
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

​

Операции, которые остаются ленивыми

Операция	Возвращает	Описание
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

---

​

Трёхфазная модель выполнения

Операции DataStore выполняются по трёхфазной модели:

​

Фаза 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

config.set_execution_engine('chdb')
# Все операции используют ClickHouse SQL

​

Принудительное использование движка pandas

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. Используйте кэширование с умом - Понимайте, когда кэш сбрасывается