Načítání dat do ClickHouse: dávky, async_insert, monitorování

Načítání dat do ClickHouse: jak jsem přestal vkládat po jednom řádku a zrychlil příjem 500krát

Měl jsem projekt – analýzu sázek v reálném čase. Denně přicházelo 50 milionů událostí. Rozhodl jsem se, že INSERT INTO bets VALUES (...) po jedné události je v pořádku. ClickHouse zvládal 500 vložení za sekundu, ale my potřebovali 2000. Server se zhroutil: diskové fronty letěly do kosmu a slučování na pozadí vytvořilo tisíce malých částí. Produkce spadla.

Ukázalo se, že ClickHouse není MySQL. Je stvořený pro hromadné vkládání, ne pro jednotlivé INSERT. Když jsem přepsal loader na dávky po 100 000 záznamech, dosáhl jsem 10 000 vložení za sekundu bez ztráty výkonu.

Níže jsou všechny způsoby načítání, od malých CSV až po produkční batche, s nástrahami, které mě stály nervy.

1. Proč je jeden řádek zlo (i když se vám chce)

ClickHouse pro každý INSERT vytvoří nový díl na disku. Díly obsahují granule po 8192 řádcích, ale i když vložíte 1 řádek – fyzicky se vytvoří nový díl s malými .bin soubory.

Důsledky:

• 10 000 INSERT po 1 řádku = 10 000 dílů
• Dotaz SELECT musí otevřít 10 000 souborů
• Slučování na pozadí se je snaží slepit, zatěžuje CPU
• Místo na disku se plýtvá na metadata dílů

Pravidlo z mé zkušenosti: dávka by měla mít alespoň 1000 řádků nebo 1 MB komprimovaných dat.

2. INSERT ... VALUES – pro testy, ne pro produkci

-- Pouze pro lokální experimenty
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');

V produkci to nelze dělat. I když batchujete po 1000 řádcích v VALUES – parsování SQL se stane úzkým hrdlem.

3. INSERT ... FORMAT CSV / JSONEachRow / TSV – záchrana pro soubory

ClickHouse rozumí několika formátům za běhu.

Příklad s CSV souborem 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

Načítáme:

INSERT INTO betting.bets FORMAT CSV

Ale data je třeba dodat na stdin. Proto se ve skriptech používá:

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

JSONEachRow – ideální pro logy:

{"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"}

Načítání:

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

TabSeparated – nejrychlejší (méně parsování):

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

4. Načítání přes clickhouse-client ze souboru – můj oblíbený způsob pro zálohy

# Malý soubor
clickhouse-client --query="INSERT INTO betting.bets FORMAT CSV" < /tmp/daily_bets.csv

# Obrovský soubor s průběhem
clickhouse-client \
  --query="INSERT INTO betting.bets FORMAT CSV" \
  --progress \
  --max_insert_block_size=100000 \
  < /data/historical_bets_2023.csv

Důležitý parametr: --max_insert_block_size=100000 – rozdělí soubor na bloky po 100k řádcích. To umožňuje vkládat terabajtové soubory bez přetečení paměti.

Co jsem si odnesl: FORMAT definuje strukturu vstupních dat, nikoli výstupních. Všimněte si – dotaz sám o sobě neobsahuje data, přicházejí přes stdin.

5. HTTP API s POST – když není přístup k příkazové řádce

# CSV přes 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 za běhu (úspora přenosu)
gzip -c bets.csv | curl -X POST "http://localhost:8123/?query=INSERT+INTO+betting.bets+FORMAT+CSV" \
  --data-binary @- \
  --header "Content-Encoding: gzip"

Proč POST, a ne GET: GET dotazy se logují kompletně, včetně dat. Pokud vkládáte přes ?query=INSERT..., heslo a data se mohou objevit v logách proxy.

6. INSERT SELECT – přesouvání dat uvnitř ClickHouse

Nejrychlejší způsob hromadného přesunu – neexportovat ven.

-- Zkopírovat data za leden do archivní tabulky
INSERT INTO betting.bets_archive
SELECT * FROM betting.bets
WHERE created_at >= '2024-01-01' AND created_at < '2024-02-01';

-- Agregovat za běhu
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;

Rychlost: INSERT SELECT probíhá bez komunikace klient-server, pouze lokální přesun. Miliarda řádků se přesune během minut.

7. Async insert: když tisíce událostí za sekundu od hráčů

Klasický INSERT je synchronní: klient čeká, dokud se data nezapíší na disk. Při 10 000 vloženích za sekundu to bolí.

async_insert bufferuje data v paměti a vkládá je v dávkách.

Zapíná se v config.xml nebo přes SETTINGS:

-- Na úrovni session
SET async_insert = 1;
SET wait_for_async_insert = 0;  -- nečekat na dokončení
SET async_insert_max_data_size = 10000000;  -- 10 MB buffer
SET async_insert_busy_timeout_ms = 200;     -- flush každých 200 ms

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

Jak to funguje:

1. Klient posílá malé vložení
2. ClickHouse je ukládá do bufferu v paměti
3. Když je buffer plný (10 MB) nebo vyprší timeout (200 ms), provede se jedno hromadné vložení

Na čem jsem se spálil: wait_for_async_insert=0 znamená, že klient neví, jestli vložení selhalo. Při síťových problémech můžete přijít o data. Nastavte wait_for_async_insert=1, pokud je důležitá spolehlivost. Výkon klesne, ale ne katastrofálně.

8. Nastavení max_insert_block_size a max_batch_size

<!-- config.xml -->
<max_insert_block_size>1048576</max_insert_block_size>  <!-- 1M řádků -->
<max_block_size>65536</max_block_size>

max_insert_block_size – maximální velikost bloku, který ClickHouse zpracuje najednou. Pokud je vaše vložení větší, je rozděleno na bloky.

Pravidlo z produkce: nastavte max_insert_block_size na 2–4krát menší než očekávaná dávka. Pak ani náhodně obrovský INSERT nezabije paměť.

# Vynuceně pro jedno vložení
clickhouse-client --query="INSERT INTO bets FORMAT CSV" \
  --max_insert_block_size=500000 < huge_file.csv

9. Monitorování vložení přes system.query_log

Nikdy nehádejte, kolik řádků se vložilo. Sledujte system.query_log.

-- Nedávné INSERT dotazy
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;

Můj dashboard pro monitoring: zobrazuji metriky za poslední hodinu – průměrnou velikost dávky, počet neúspěšných vložení, rychlost řádků/s.

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. Python skript pro dávkové načítání historických dat

V reálném projektu jsme nahrávali 3 roky historie sázek z parquetů v S3. Toto je skript, který zvládl 2 miliardy řádků:

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

# Připojení k ClickHouse
client = Client(
    host='clickhouse.prod.internal',
    port=9000,
    user='loader',
    password='strong_password',
    database='betting'
)

# Optimalizace pro velká vložení
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'):
    """Vložení DataFrame dávkou přes Native protokol"""
    # ClickHouse-driver sám konvertuje typy
    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="Zpracování souborů"):
        # Čteme parquet po částech, abychom neudrželi vše v paměti
        for chunk in pd.read_parquet(file, chunksize=batch_size):
            # Převod typů pro 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 částí po 500k = 2.5M řádků
                df_batch = pd.concat(buffer, ignore_index=True)
                insert_batch(df_batch)
                buffer = []
    
    # Zbývající data
    if buffer:
        df_batch = pd.concat(buffer, ignore_index=True)
        insert_batch(df_batch)

if __name__ == '__main__':
    stream_parquet_files()

Alternativa přes requests (bez 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"Vložení selhalo: {response.text}")
    
    print(f"Vloženo {filepath}")

# Příklad dávkového odeslání více souborů
for file in glob.glob('/data/bets_*.csv.gz'):
    insert_via_http_csv_gz(file)

Typické chyby při načítání (a jak jsem je opravil)

Chyba: Memory limit (total) exceeded
Příčina: příliš velký blok vložení.
Řešení: zmenšit max_insert_block_size nebo rozdělit soubor na kusy.

Chyba: Too many parts
Příčina: příliš mnoho malých INSERT nebo vypnutý async_insert při vysokém toku.
Řešení: zapnout async_insert, zvýšit min_rows_for_wide_part.

Chyba: Code: 252. DB::Exception: Too many simultaneous inserts
Příčina: paralelní vložení do stejné partition.
Řešení: nastavit max_insert_threads = 1 pro tuto tabulku.

Chyba: partition se neslučují
Příčina: data se vkládají mimo pořadí ORDER BY.
Řešení: ujistěte se, že data jsou seřazena stejně jako ORDER BY v tabulce.

Co dál

Umíte načítat data jakýmkoli způsobem – od malých CSV po produkční toky. Další článek bude o optimalizaci dotazů a pokročilých agregacích.

Hlavní rada na rozloučenou: žádný INSERT by neměl opustit váš skript bez páru max_insert_block_size a async_insert, pokud pracujete s produkcí. Na tyhle hrábě jsem šlápl třikrát – dost.

---

← Předchozí: MergeTree v ClickHouse: jak engine řeže analytiku na granule a slučuje části
→ Další: SELECT dotazy v ClickHouse: jak jsem přeučoval mozek po 10 letech PostgreSQL

— Editorial Team