Carga de datos en ClickHouse: lotes, async_insert, monitoreo

Carga de datos en ClickHouse: Cómo dejé de insertar una fila a la vez y aceleré la ingesta 500 veces

Una historia de desastre: 1000 consultas por segundo y una caída en producción

Tenía un proyecto: análisis de apuestas en tiempo real. Llegaban 50 millones de eventos al día. Pensé que INSERT INTO bets VALUES (...) un evento a la vez estaba bien. ClickHouse manejaba 500 inserciones por segundo, pero necesitábamos 2000. El servidor colapsó: las colas de disco se dispararon y las fusiones en segundo plano generaron miles de partes diminutas. La producción se cayó.

Resulta que ClickHouse no es MySQL. Está diseñado para inserciones masivas, no para INSERT puntuales. Después de reescribir el cargador para agrupar 100 000 registros a la vez, vi 10 000 inserciones por segundo sin pérdida de rendimiento.

A continuación, todos los métodos de carga, desde pequeños CSV hasta cargadores de nivel de producción, junto con los errores que me costaron noches sin dormir.

1. Por qué una sola fila es mala (incluso si quieres)

ClickHouse crea una nueva parte en disco por cada INSERT. Las partes contienen gránulos de 8192 filas, pero incluso si insertas solo 1 fila, se crea físicamente una nueva parte con pequeños archivos .bin.

Consecuencias:

• 10 000 INSERT de 1 fila cada uno = 10 000 partes
• Una consulta SELECT debe abrir 10 000 archivos
• Las fusiones en segundo plano intentan combinarlos, sobrecargando la CPU
• Se desperdicia espacio en disco con metadatos de partes

Regla de mi experiencia: un lote debe tener al menos 1000 filas o 1 MB de datos comprimidos.

2. INSERT ... VALUES — Para pruebas, no para producción

-- Solo para experimentos locales
INSERT INTO betting.bets (user_id, created_at, amount, odds, sport, outcome)
VALUES 
    (1001, now(), 50.00, 2.1, 'football', 'win'),
    (1002, now(), 100.00, 1.8, 'basketball', 'loss'),
    (1003, now(), 75.00, 3.0, 'tennis', 'win');

En producción, nunca hagas esto. Incluso si agrupas 1000 filas en VALUES, el análisis sintáctico de SQL se convierte en el cuello de botella.

3. INSERT ... FORMAT CSV / JSONEachRow / TSV — Salvavidas para archivos

ClickHouse entiende varios formatos sobre la marcha.

Ejemplo con archivo CSV bets.csv:

1001,2024-03-15 10:00:00,50.00,2.1,football,win
1002,2024-03-15 10:01:00,100.00,1.8,basketball,loss
1003,2024-03-15 10:02:00,75.00,3.0,tennis,win

Cárgalo:

INSERT INTO betting.bets FORMAT CSV

Pero necesitas canalizar los datos a stdin. Así que en scripts, usa:

clickhouse-client --query="INSERT INTO betting.bets FORMAT CSV" < bets.csv

JSONEachRow — ideal para logs:

{"user_id":1001,"created_at":"2024-03-15 10:00:00","amount":50.00,"odds":2.1,"sport":"football","outcome":"win"}
{"user_id":1002,"created_at":"2024-03-15 10:01:00","amount":100.00,"odds":1.8,"sport":"basketball","outcome":"loss"}

Carga:

clickhouse-client --query="INSERT INTO betting.bets FORMAT JSONEachRow" < bets.json

TabSeparated — el más rápido (menos análisis):

clickhouse-client --query="INSERT INTO betting.bets FORMAT TSV" < bets.tsv

4. Carga mediante clickhouse-client desde archivo — Mi favorita para copias de seguridad

# Archivo pequeño
clickhouse-client --query="INSERT INTO betting.bets FORMAT CSV" < /tmp/daily_bets.csv

# Archivo enorme con progreso
clickhouse-client \
  --query="INSERT INTO betting.bets FORMAT CSV" \
  --progress \
  --max_insert_block_size=100000 \
  < /data/historical_bets_2023.csv

Indicador importante: --max_insert_block_size=100000 — divide el archivo en bloques de 100k filas. Esto permite insertar archivos de tamaño terabyte sin desbordar la memoria.

Lo que aprendí: FORMAT define la estructura de los datos de entrada, no la salida. Observa que la consulta en sí no contiene datos; estos vienen a través de stdin.

5. API HTTP con POST — Cuando no tienes acceso a la línea de comandos

# CSV mediante curl
curl -X POST "http://localhost:8123/?query=INSERT+INTO+betting.bets+FORMAT+CSV" \
  --data-binary @bets.csv

# JSONEachRow
curl -X POST "http://localhost:8123/?query=INSERT+INTO+betting.bets+FORMAT+JSONEachRow" \
  --data-binary @bets.json \
  -u analyst:password

# GZIP sobre la marcha (ahorra ancho de banda)
gzip -c bets.csv | curl -X POST "http://localhost:8123/?query=INSERT+INTO+betting.bets+FORMAT+CSV" \
  --data-binary @- \
  --header "Content-Encoding: gzip"

¿Por qué POST y no GET? Las solicitudes GET se registran completamente, incluidos los datos. Si insertas mediante ?query=INSERT..., la contraseña y los datos pueden terminar en los logs del proxy.

6. INSERT SELECT — Mover datos dentro de ClickHouse

La forma más rápida de recargar en masa no es exportar externamente.

-- Copiar datos de enero a la tabla de archivo
INSERT INTO betting.bets_archive
SELECT * FROM betting.bets
WHERE created_at >= '2024-01-01' AND created_at < '2024-02-01';

-- Agregar sobre la marcha
INSERT INTO betting.daily_stats (date, total_bets, total_amount)
SELECT 
    toDate(created_at) AS date,
    count() AS total_bets,
    sum(amount) AS total_amount
FROM betting.bets
WHERE created_at >= today() - 7
GROUP BY date;

Velocidad: INSERT SELECT evita el intercambio cliente-servidor; es una transferencia local. Se pueden mover mil millones de filas en minutos.

7. Inserción asíncrona: Cuando llegan miles de eventos por segundo desde los jugadores

El INSERT clásico es síncrono: el cliente espera hasta que los datos se escriben en disco. Con 10 000 inserciones por segundo, eso duele.

async_insert almacena datos en búfer en memoria e inserta en lotes.

Actívalo en config.xml o mediante SETTINGS:

-- Nivel de sesión
SET async_insert = 1;
SET wait_for_async_insert = 0;  -- no esperar a que termine
SET async_insert_max_data_size = 10000000;  -- búfer de 10 MB
SET async_insert_busy_timeout_ms = 200;     -- vaciar cada 200 ms

INSERT INTO betting.bets FORMAT JSONEachRow
{"user_id":1001,"amount":50,"odds":2.1}
{"user_id":1002,"amount":100,"odds":1.8}

Cómo funciona:

1. El cliente envía inserciones pequeñas
2. ClickHouse las acumula en un búfer en memoria
3. Cuando el búfer está lleno (10 MB) o el tiempo de espera expira (200 ms), se produce una única inserción masiva

Lo que me quemó: wait_for_async_insert=0 significa que el cliente no sabe si la inserción falló. Los problemas de red pueden causar pérdida de datos. Establece wait_for_async_insert=1 si la fiabilidad importa. El rendimiento baja, pero no catastróficamente.

8. Ajustes: max_insert_block_size y max_batch_size

<!-- config.xml -->
<max_insert_block_size>1048576</max_insert_block_size>  <!-- 1M filas -->
<max_block_size>65536</max_block_size>

max_insert_block_size — el tamaño máximo de bloque que ClickHouse procesa a la vez. Si tu inserción es más grande, se divide en bloques.

Regla de producción: establece max_insert_block_size de 2 a 4 veces más pequeño que tu lote esperado. De esa manera, incluso un INSERT accidentalmente enorme no matará la memoria.

# Forzar para una sola inserción
clickhouse-client --query="INSERT INTO bets FORMAT CSV" \
  --max_insert_block_size=500000 < huge_file.csv

9. Monitoreo de inserciones mediante system.query_log

Nunca adivines cuántas filas se insertaron. Consulta system.query_log.

-- Consultas INSERT recientes
SELECT 
    event_time,
    query_duration_ms,
    read_rows,
    written_rows,
    result_rows,
    memory_usage,
    query
FROM system.query_log
WHERE type = 'QueryFinish'
  AND query LIKE '%INSERT INTO betting.bets%'
  AND event_time >= now() - INTERVAL 1 HOUR
ORDER BY event_time DESC
LIMIT 20;

Mi panel de monitoreo: muestra métricas de la última hora: tamaño promedio del lote, número de inserciones fallidas, velocidad de filas/segundo.

SELECT 
    toStartOfMinute(event_time) AS minute,
    count() AS inserts,
    avg(written_rows) AS avg_batch_size,
    sum(written_rows) AS total_rows,
    avg(query_duration_ms) AS avg_latency_ms
FROM system.query_log
WHERE type = 'QueryFinish'
  AND query LIKE '%INSERT INTO betting.bets%'
  AND event_time >= now() - INTERVAL 1 HOUR
GROUP BY minute
ORDER BY minute DESC;

10. Script en Python para carga por lotes de datos históricos

En un proyecto real, cargamos 3 años de historial de apuestas desde archivos parquet en S3. Aquí hay un script que manejó 2 mil millones de filas:

#!/usr/bin/env python3
from clickhouse_driver import Client
import pandas as pd
import glob
from tqdm import tqdm

# Conectar a ClickHouse
client = Client(
    host='clickhouse.prod.internal',
    port=9000,
    user='loader',
    password='strong_password',
    database='betting'
)

# Optimizaciones para inserciones grandes
client.execute("SET max_insert_block_size = 1000000")
client.execute("SET min_insert_block_size_rows = 500000")
client.execute("SET min_insert_block_size_bytes = 100000000")  # 100 MB

def insert_batch(df, table='bets'):
    """Insertar DataFrame como un lote mediante el protocolo Native"""
    # clickhouse-driver convierte tipos automáticamente
    client.execute(f"INSERT INTO {table} FORMAT TabSeparated", 
                   df.to_csv(sep='\t', header=False, index=False),
                   types_check=True)

def stream_parquet_files(pattern='/data/bets_*.parquet'):
    files = glob.glob(pattern)
    batch_size = 500000
    buffer = []
    
    for file in tqdm(files, desc="Processing files"):
        # Leer parquet en fragmentos para no mantener todo en memoria
        for chunk in pd.read_parquet(file, chunksize=batch_size):
            # Convertir tipos para ClickHouse
            chunk['created_at'] = pd.to_datetime(chunk['created_at'])
            chunk['amount'] = chunk['amount'].astype('float64')
            chunk['odds'] = chunk['odds'].astype('float64')
            
            buffer.append(chunk)
            if len(buffer) >= 5:  # 5 fragmentos de 500k = 2.5M filas
                df_batch = pd.concat(buffer, ignore_index=True)
                insert_batch(df_batch)
                buffer = []
    
    # Datos restantes
    if buffer:
        df_batch = pd.concat(buffer, ignore_index=True)
        insert_batch(df_batch)

if __name__ == '__main__':
    stream_parquet_files()

Alternativa mediante requests (sin pandas):

import requests
import gzip
import io

def insert_via_http_csv_gz(filepath):
    with open(filepath, 'rb') as f:
        compressed = gzip.compress(f.read())
    
    response = requests.post(
        'http://localhost:8123/?query=INSERT+INTO+betting.bets+FORMAT+CSV',
        data=compressed,
        headers={'Content-Encoding': 'gzip'},
        auth=('loader', 'password')
    )
    
    if response.status_code != 200:
        raise Exception(f"Insert failed: {response.text}")
    
    print(f"Inserted {filepath}")

# Ejemplo de envío por lotes de múltiples archivos
for file in glob.glob('/data/bets_*.csv.gz'):
    insert_via_http_csv_gz(file)

Errores comunes de carga (y cómo los solucioné)

Error: Memory limit (total) exceeded
Causa: Bloque de inserción demasiado grande.
Solución: Reducir max_insert_block_size o dividir el archivo en partes.

Error: Too many parts
Causa: Demasiados INSERT pequeños o async_insert deshabilitado bajo alto rendimiento.
Solución: Habilitar async_insert, aumentar min_rows_for_wide_part.

Error: Code: 252. DB::Exception: Too many simultaneous inserts
Causa: Inserciones paralelas en la misma partición.
Solución: Establecer max_insert_threads = 1 para esa tabla.

Error: Las particiones no se fusionan
Causa: Datos insertados fuera del orden ORDER BY.
Solución: Asegurar que los datos estén ordenados igual que el ORDER BY de la tabla.

Qué sigue

Ahora sabes cómo cargar datos por cualquier método — desde pequeños CSV hasta flujos de producción. El próximo artículo cubre optimización de consultas y agregaciones avanzadas.

Consejo de despedida: Ningún INSERT debería salir de tu script sin un par de max_insert_block_size y async_insert si trabajas con producción. Pisé ese rastrillo tres veces — suficiente.

---

← Anterior: MergeTree en ClickHouse: Cómo el motor divide la analítica en gránulos y fusiona partes
→ Siguiente: Consultas SELECT en ClickHouse: Cómo reentrené mi cerebro después de 10 años con PostgreSQL

— Editorial Team