Modelo de ejecución de DataStore

Comprender la evaluación diferida, qué desencadena la ejecución y el almacenamiento en caché en DataStore

Comprender el modelo de evaluación diferida de DataStore es fundamental para usarlo eficazmente y obtener un rendimiento óptimo.

​

Evaluación diferida

DataStore usa evaluación diferida: las operaciones no se ejecutan de inmediato, sino que se registran y se compilan en consultas SQL optimizadas. La ejecución solo se produce cuando realmente se necesitan los resultados.

​

Ejemplo: evaluación diferida vs. inmediata

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

# Estas operaciones NO se ejecutan todavía
result = (ds
    .filter(ds['amount'] > 1000)    # Registrado, no ejecutado
    .select('region', 'amount')      # Registrado, no ejecutado
    .groupby('region')               # Registrado, no ejecutado
    .agg({'amount': 'sum'})          # Registrado, no ejecutado
    .sort('sum', ascending=False)    # Registrado, no ejecutado
)

# Aún sin ejecución - solo construyendo el plan de consulta
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

# AHORA ocurre la ejecución
df = result.to_df()  # <-- Dispara la ejecución

​

Beneficios de la evaluación diferida

1. Optimización de consultas: Varias operaciones se convierten en una única consulta SQL optimizada
2. Pushdown de filtros: Los filtros se aplican a nivel de la fuente de datos
3. Poda de columnas: Solo se leen las columnas necesarias
4. Decisiones diferidas: El motor de ejecución puede elegirse en tiempo de ejecución
5. Inspección del plan: Puede ver/depurar la consulta antes de ejecutarla

---

​

Activadores de ejecución

La ejecución se activa automáticamente cuando se necesitan valores reales:

​

Activadores automáticos

Activador	Ejemplo	Descripción
print() / repr()	print(ds)	Mostrar resultados
len()	len(ds)	Obtener el número de filas
.columns	ds.columns	Obtener los nombres de las columnas
.dtypes	ds.dtypes	Obtener los tipos de las columnas
.shape	ds.shape	Obtener las dimensiones
.index	ds.index	Obtener el índice de filas
.values	ds.values	Obtener un array de NumPy
Iteración	for row in ds	Iterar sobre las filas
to_df()	ds.to_df()	Convertir a pandas
to_pandas()	ds.to_pandas()	Alias de to_df
to_dict()	ds.to_dict()	Convertir a dict
to_numpy()	ds.to_numpy()	Convertir a un array
.equals()	ds.equals(other)	Comparar DataStores

Ejemplos:

# Todos estos activan la ejecución
print(ds)              # Mostrar
len(ds)                # 1000
ds.columns             # Index(['name', 'age', 'city'])
ds.shape               # (1000, 3)
list(ds)               # Lista de valores
ds.to_df()             # pandas DataFrame

​

Operaciones que siguen siendo diferidas

Operación	Devuelve	Descripción
filter()	DataStore	Añade la cláusula WHERE
select()	DataStore	Añade la selección de columnas
sort()	DataStore	Añade ORDER BY
groupby()	LazyGroupBy	Prepara GROUP BY
join()	DataStore	Añade JOIN
ds['col']	ColumnExpr	Referencia de columna
ds[['col1', 'col2']]	DataStore	Selección de columnas

Ejemplos:

# Estas operaciones NO activan la ejecución - permanecen en modo lazy
result = ds.filter(ds['age'] > 25)      # Devuelve DataStore
result = ds.select('name', 'age')        # Devuelve DataStore
result = ds['name']                      # Devuelve ColumnExpr
result = ds.groupby('city')              # Devuelve LazyGroupBy

---

​

Ejecución en tres fases

Las operaciones de DataStore siguen un modelo de ejecución de tres fases:

​

Fase 1: Construcción perezosa de consultas SQL

Se acumulan las operaciones que pueden expresarse en 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
)
# Todo esto se compila en una única consulta SQL

​

Fase 2: Punto de ejecución

Cuando ocurre un evento activador, se ejecuta el SQL acumulado:

# La ejecución se activa aquí
df = result.to_df()  
# La consulta SQL optimizada se ejecuta ahora

​

Fase 3: Operaciones de DataFrame (si las hay)

Si encadenas operaciones solo de pandas después de la ejecución:

# Operaciones mixtas
result = (ds
    .filter(ds['amount'] > 100)          # Fase 1: SQL
    .to_df()                             # Fase 2: Ejecutar
    .pivot_table(...)                    # Fase 3: pandas
)

---

​

Ver planes de ejecución

Utiliza explain() para ver qué se ejecutará:

Query

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

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

# Ver el plan de ejecución
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

Usa verbose=True para obtener más información:

query.explain(verbose=True)

Consulte Depuración: explain() para obtener la documentación completa.

---

​

Caché

DataStore guarda en caché los resultados de ejecución para evitar consultas redundantes.

​

Cómo funciona la caché

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)

# Primer acceso - ejecuta la consulta
print(result.shape)  # Ejecuta y almacena en caché

# Segundo acceso - usa la caché
print(result.columns)  # Usa el resultado almacenado en caché

# Tercer acceso - usa la caché
df = result.to_df()  # Usa el resultado almacenado en caché

​

Invalidación de caché

La caché se invalida cuando se realizan operaciones que modifican el DataStore:

result = ds.filter(ds['age'] > 25)
print(result.shape)  # Ejecuta, almacena en caché

# Una nueva operación invalida el caché
result2 = result.filter(result['city'] == 'NYC')
print(result2.shape)  # Re-ejecuta (consulta diferente)

​

Control manual de caché

# Limpiar caché
ds.clear_cache()

# Deshabilitar el almacenamiento en caché
from chdb.datastore.config import config
config.set_cache_enabled(False)

---

​

Combinación de operaciones SQL y de Pandas

DataStore gestiona de forma inteligente las operaciones que combinan SQL y Pandas:

​

Operaciones compatibles con SQL

Se compilan a SQL:

• filter(), where()
• select()
• groupby(), agg()
• sort(), orderby()
• limit(), offset()
• join(), union()
• distinct()
• Operaciones sobre columnas (matemáticas, comparación, métodos de cadenas)

​

Operaciones exclusivas de Pandas

Estas operaciones desencadenan la ejecución y usan Pandas:

• apply() con funciones personalizadas
• pivot_table() con agregaciones complejas
• stack(), unstack()
• Operaciones sobre DataFrames ejecutados

​

Canalizaciones híbridas

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

# Fase de ejecución + pandas
result = (result
    .to_df()                          # Ejecutar SQL
    .pivot_table(...)                 # operación de pandas
)

---

​

Selección del motor de ejecución

DataStore puede ejecutar operaciones con distintos motores:

​

Modo automático (predeterminado)

from chdb.datastore.config import config

config.set_execution_engine('auto')  # Predeterminado
# Selecciona automáticamente el mejor motor por operación

​

Forzar el motor chDB

config.set_execution_engine('chdb')
# Todas las operaciones usan ClickHouse SQL

​

Forzar el motor de pandas

config.set_execution_engine('pandas')
# Todas las operaciones usan pandas

Consulta Configuración: motor de ejecución para más información.

---

​

Implicaciones en el rendimiento

​

Bien: filtrar pronto

# Correcto: filtrar en SQL, luego agregar
result = (ds
    .filter(ds['date'] >= '2024-01-01')  # Reduce los datos con anticipación
    .groupby('category')
    .agg({'amount': 'sum'})
)

​

Mal: filtrar tarde

# Incorrecto: Agregar todo y luego filtrar
result = (ds
    .groupby('category')
    .agg({'amount': 'sum'})
    .to_df()
    .query('sum > 1000')  # Filtro de Pandas después de la agregación
)

​

Bien: Selecciona las columnas desde el principio

# Correcto: Seleccionar columnas en SQL
result = (ds
    .select('user_id', 'amount', 'date')
    .filter(ds['date'] >= '2024-01-01')
    .groupby('user_id')
    .agg({'amount': 'sum'})
)

​

Bien: deja que SQL haga el trabajo

# Bien: Agregación compleja en SQL
result = (ds
    .groupby('category')
    .agg({
        'amount': ['sum', 'mean', 'count'],
        'quantity': 'sum'
    })
    .sort('sum', ascending=False)
    .limit(10)
)
# Una sola consulta SQL hace todo

# Mal: Múltiples consultas separadas
sums = ds.groupby('category')['amount'].sum().to_df()
means = ds.groupby('category')['amount'].mean().to_df()
# Dos consultas en lugar de una

---

​

Resumen de buenas prácticas

1. Encadena las operaciones antes de ejecutar - Construye la consulta completa y luego ejecútala una sola vez
2. Filtra cuanto antes - Reduce los datos en el origen
3. Selecciona solo las columnas necesarias - Descartar columnas mejora el rendimiento
4. Usa explain() para entender la ejecución - Depura antes de ejecutar
5. Deja que SQL se encargue de las agregaciones - ClickHouse está optimizado para ello
6. Ten en cuenta qué desencadena la ejecución - Evita ejecutar antes de tiempo por accidente
7. Usa la caché con criterio - Comprende cuándo se invalida