5 jaar ervaring met ClickHouse®-clusters op petabyte-schaal

Wat ik heb geleerd bij het beheren van ClickHouse op grote schaal: de successen, de fouten en de lessen die alleen voortkomen uit praktijkervaring in productie.

Ik beheer bij Tinybird een ClickHouse®-cluster sinds versie 18.4, bijna zes jaar geleden. Bijna acht jaar geleden schreef ik mijn eerste blogpost over hoe ClickHouse® kan worden gebruikt voor geospatial analyse.

In die tijd heb ik dagelijks met ClickHouse® gewerkt, geholpen bij het oprichten van een bedrijf dat het gebruikt, kritieke wijzigingen naar het databaseproject gestuurd en vele clusters op petabyte-schaal beheerd. Deze review deelt de geleerde lessen. Het opzetten van een cluster is eenvoudig; het moeilijke gedeelte is het draaiende houden. Ik zal ingaan op de goede en de slechte kanten, met een focus op de problemen die je kunt tegenkomen (zodat je deze kunt vermijden).

Deze review is waarschijnlijk nuttig voor mensen die ClickHouse®-clusters beheren, maar minder voor anderen.

Belangrijke opmerking, hoofdzakelijk voor de juristen van ClickHouse®, Inc.: wij hebben niets te maken met ClickHouse®, Inc. Zij zijn de enige beheerders van ClickHouse®, en ClickHouse® is een geregistreerd handelsmerk in eigendom van ClickHouse®, Inc. Mijn excuses voor eventuele fouten in deze tekst; ik sta open voor correcties. Wij zijn slechts kleine bijdragers van nieuwe engines, enorme prestatieverbeteringen, distributed joins en andere zaken, en we zijn uiteraard zelf gebruikers van de open-source versie. Ik heb groot respect voor Alexey voor het starten van het project en voor de vele mensen die daar werken.

Architectuur

De architectuur die ClickHouse® voorstelt voor het beheren van grote clusters maakt gebruik van replica's en shards. In essentie splits je je gegevens in verschillende 'buckets' (bijv. per hash(user_id)) en elke bucket gaat naar een shard. Vervolgens heb je verschillende replica's (kopieën van de data) voor elke shard.

Dit is een vrij standaard en eenvoudige architectuur, maar het begint aan te voelen als de verkeerde benadering voor dit soort systemen. Velen geloven nu dat cloud-opslag gebruikt moet worden om compute (rekenkracht) te scheiden van storage (opslag). Ik ben het daar niet volledig mee eens, maar deze benadering biedt wel voordelen voor clusterbeheer en kostenvermindering. Ik kom hier later uitgebreider op terug.

Een paar jaar geleden begonnen we met een vrij basis systeem zonder shards, alleen replica's. We schaalden de replica's verticaal om grotere queries af te handelen en voegden meer replica's toe om meer verkeer te verwerken. Cloud-opslag was geen optie, dus gebruikten we lokale SSD-schijven om de latentie te verlagen. Het toevoegen van shards was een optie geweest als we meer data per query moesten verwerken, maar we hebben dat uiteindelijk niet gedaan omdat re-sharding extreem moeilijk was en geen van onze klanten daar behoefte aan had (of liever gezegd: als je slim genoeg bent in je data-schema ontwerp, kun je dit uitstellen).

Die configuratie is vrij eenvoudig te beheren; je plaatst een loadbalancer voor alle replica's en routeert het verkeer van je applicatie naar een replica, afhankelijk van het type verzoek. We gebruiken een HTTP-loadbalancer met logica om instructies van de backend af te handelen. De backend (de applicaties in het diagram) neemt beslissingen over waar een specifiek verzoek naartoe moet op basis van belasting, replica-type, consistent hashing (om caches te benutten, zie Opslag) en meer. Ik denk niet dat je dit allemaal nodig hebt voor een basisinstallatie, maar een deel ervan is essentieel.

Een korte opmerking over HTTP: ClickHouse® biedt een TCP-connector met een native protocol, maar die gebruiken we niet. Het biedt niet veel voordelen voor het type applicaties dat wij bouwen, en HTTP stelt ons in staat om veel beproefde tooling te gebruiken. Als je verbindt vanuit applicatiecode, kun je taalspecifieke clients gebruiken (zoals Python, Java of Go) die beide protocollen ondersteunen.

Deze architectuur kan erg duur worden, vooral omdat je een replica van alle data op alle machines nodig hebt (afhankelijk van hoe je loadbalancing en beschikbaarheid inricht). Stel dat je een tabel van 300TB hebt en 1000 QPS (queries per seconde) moet afhandelen. Als elke replica 100 QPS aankan, heb je 10 replica's nodig, wat neerkomt op 300 * 10 = 3000TB. Als je SSD's gebruikt, heb je een probleem.

Ik raad altijd aan om een replica uitsluitend voor writes (schrijfacties) te hebben; dit wordt compute-compute scheiding genoemd. Wij doen dit standaard en beheren de failover en andere zaken in geval van fouten of overbelasting.

Je kunt er ook voor kiezen om het cluster op te splitsen in meer replica's, afhankelijk van de verschillende workloads. Als bepaalde queries bijvoorbeeld een specifiek p99-doel hebben, wil je die queries wellicht isoleren naar een replica en deze onder de 40% belasting houden, zodat de hoge percentielen stabiel blijven. Real-time sub-seconde queries zijn duur; vergeet die goedkope batch-jobs die op spot-instances draaien.

De loadbalancer is de sleutel tot dit alles. Elke moderne LB is hiervoor geschikt.

Opslag

Open-source ClickHouse® staat voor een aanzienlijke uitdaging: beperkte ondersteuning voor cloud-opslag. Moderne OLAP-databases en datasystemen zouden cloud-opslag moeten benutten voor kostenefficiëntie en onafhankelijke schaling van compute- en storage-resources. Snowflake stelde deze standaard meer dan tien jaar geleden vast, en ClickHouse® (open source) loopt hierop achter, terwijl andere systemen zoals StarRocks veel verder zijn.

Hier is een korte uitleg over hoe opslag werkt in ClickHouse® (en andere databases). Zoals gezegd kun je je data opsplitsen in "buckets" en elke bucket in een shard plaatsen. Elke shard kan één of meer replica's (kopieën) van de data hebben.

1. Lokale opslag: De data wordt lokaal opgeslagen, waardoor elke replica alle data van andere replica's moet kopiëren. Je voert data in bij één shard, en de data wordt naar anderen gerepliceerd. In ClickHouse® wordt dit beheerd door een centraal systeem genaamd Zookeeper.
2. Centrale opslag: De data kan worden opgeslagen in een centraal systeem, zoals S3, waardoor replica's de data daar kunnen opslaan in plaats van op een lokale schijf. Elke replica kan een kopie opslaan of simpelweg verwijzen naar data van een andere replica (zero-copy replication).

Er zijn enkele gevallen waarin cloud-opslag geen zin heeft, maar in de meeste gevallen is de scheiding tussen compute en storage de juiste weg. Wat ik hier beschrijf, maakt gebruik van cloud-opslag, tenzij anders vermeld.

In ClickHouse® heb je al deze opties: lokale schijf, cloud-storage en het zero-copy replicatiemechanisme dat vertrouwt op cloud-opslag.

Die laatste, zero-copy replicatie, is bijgedragen door iemand van buiten ClickHouse®, Inc., en het lijkt erop dat zij er niet dol op zijn. Ze hebben goede redenen: het is buggy, je kunt data verliezen, het laat 'vuil' achter in S3, etc. ClickHouse® Cloud heeft zijn eigen opslag, maar die is niet open source. Toch is zero-copy, zelfs met deze beperkingen, bruikbaar als je op de hoogte bent van de tekortkomingen.

Om kostenefficiente prestaties te behalen, gebruiken we een aangepaste zero-copy functie (beschikbaar in onze private fork), maar je kunt het ook zoals het is gebruiken in de open-source repo van ClickHouse®. Onze versie bevat enkele wijzigingen, operationele beperkingen en optimalisaties. We gebruiken daarnaast lokale SSD's voor caching en passen de cachegrootte aan op basis van het gebruik. Voor klanten die lage latentie prioriteren, gebruiken we een hot/cold storage architectuur met lokale SSD's en S3. Deze aanpak biedt de lage latentie die S3 alleen niet kan leveren.

Over de kosten: De manier waarop ClickHouse® data in S3 opslaat is inefficiënt; standaard worden er veel write-operaties gebruikt. Dit komt doordat het opslagsysteem van ClickHouse® is ontworpen voor lokale opslag waar schrijfoperaties "gratis" zijn (je wordt beperkt door IOPS). Het gebruikt zeer veel bestanden om data-parts op te slaan en gebruikt de OS page cache als cache, waardoor het niet is ontworpen met tiered storage in gedachten. Er zijn echter manieren om dit te verbeteren.

Mijn algemene advies is: als je compute-storage scheiding wilt, gebruik dan zero-copy replicatie, maar houd het goed in de gaten, want je kunt data verliezen bij bepaalde part-operaties. Bovendien zou ClickHouse®, Inc. de functie op elk moment kunnen verwijderen. Als je dat risico niet wilt lopen, kies dan voor hot/cold of alleen SSD's en houd je kosten nauwlettend in de gaten.

Over compressie: Gebruik ZSTD(1) of misschien ZSTD(2) voor een goed compromis tussen snelheid en compressie. Dit is beter dan LZ4, behalve in een paar specifieke gevallen, maar je kunt de compressie per kolom instellen. Test af en toe andere compressieformaten om te zien of jouw data toevallig beter werkt met iets anders dan ZSTD.

Upgrades

Upgrades zijn nu "gemakkelijk". Nou ja, zo gemakkelijk als het updaten van een databasecluster kan zijn, wat nooit echt gemakkelijk is. Onze eerste update kostte ons 3 uur met 2 weken voorbereiding. Met de tijd leerden we hoe we dit konden doen zonder downtime, zonder applicaties te laten crashen of data te verliezen. Daarna slaagden we erin dit te doen zonder prestatieverlies en uiteindelijk als onderdeel van de CI/CD-pipeline. Het kostte ons vier jaar om dit uit te vogelen. Voor zover ik weet, is er geen ander bedrijf dat dit doet.

Mijn algemene workflow-advies voor het updaten van een cluster:

1. Voeg een nieuwe replica toe met de nieuwe versie.
2. Monitor de logs van de replica's en controleer of er iets mis is. De eerste keer zul je een paar hartaanvallen krijgen, want ClickHouse® logt graag zaken die kritiek lijken, maar dat niet zijn.
3. Vermijd indien mogelijk DDL-wijzigingen zoals het toevoegen van kolommen, data-skipping indices of alles wat de tabelstructuur wijzigt.
4. Gebruik overduidelijk geen nieuwe functies in de nieuwe replica totdat je hele cluster up-to-date is met de nieuwe versie.
5. Stuur wat leesverkeer om te controleren of alles in orde is (timing, geheugen, resourceverbruik...).
6. Test writes/inserts.
7. Verplaats nu het echte verkeer. Als je onzeker bent, houd de replica dan nog een paar uur in observatie.
8. Begin daarna met het updaten van de andere replica's.

Dit is mogelijk dankzij het backward-compatible replicatieprotocol. We hebben veel werk verricht om dit protocol backward compatible te houden. Onze CI-pipeline test upgrades, waardoor we problemen konden detecteren voordat de ClickHouse®-release plaatsvond (zodat we een PR konden sturen). Gelukkig gebeurt dit bijna niet meer; het is nu veel stabieler.

Problemen die je zult tegenkomen

• Zeldzaam: incompatibele wijzigingen in het data-opslagformaat. Dit is ons de afgelopen twee jaar 2-3 keer overkomen. We merkten het omdat we alle mogelijke combinaties van datatypes hebben in ons klantenbestand. Het is onwaarschijnlijk dat je dit tegenkomt in een reguliere (niet multi-tenant) deployment. In zulke gevallen: veel succes, bereid je voor om diep in het bestandssysteem te duiken of een backup te herstellen. Update niets direct na een release; wacht minstens een maand. Gebruik geen experimentele functies tenzij je precies weet wat je doet.
• Wijzigingen in SQL-gedrag. Dit wordt meestal veroorzaakt door bugfixes, maar het kan je queries breken. Zorg dat je de verwachte data correct test. Gebruik altijd de system.query_log met alle queries die je cluster uitvoert.
• Prestatiewijzigingen. Soms beter, soms slechter. Meestal zijn ze niet rampzalig, maar test je queries op prestaties, anders moet je terugrollen.
• Wijzigingen in instellingen: Hier moet je heel goed naar kijken. Controleer nieuwe instellingen en wijzigingen in bestaande. Let op functies die standaard zijn uitgeschakeld of flags die veranderen. De changelog is een goede start, maar het is raadzaam om je CI te laten wijzen naar de nieuwste master build.

We hebben een complex systeem om te valideren of alles in orde is vóór een upgrade:

• In CI draaien we alle tests in verschillende versies en draaien we clusters met een mix van versies (zodat we weten dat ze samen in één cluster kunnen leven). Dit zou jij ook moeten doen.
• Eén keer per dag draaien we alle queries van onze klanten met de volgende versie om te controleren of alles werkt. Als dat niet zo is, lossen we het op of informeren we de klant.
• We updaten clusters automatisch als alle tests zijn geslaagd.

Kortom: het updaten van je cluster vereist "spiergeheugen". Het gebeurt niet van de ene op de andere dag; het gebeurt door herhaling, lijden en herstel.

Configuratie en testen

Zoals eerder gezegd: houd bij welke instellingen veranderen. Bij Tinybird hebben we daar een automatisch systeem voor. Pro tip: controleer de diffs tussen versies in de broncode.

Een belangrijke opmerking: als je ClickHouse® effectief wilt beheren, moet je de broncode lezen. Dit was niet nodig bij andere databases waar ik mee werkte (ik was CTO van een bedrijf dat honderden Postgres-clusters beheerde), maar bij ClickHouse® is dit wel het geval. De meeste grote bedrijven die ClickHouse® draaien, hebben patches upstream.

Over testen: mijn aanbeveling is om je CI je applicatie te laten draaien tegen de versie die je in productie hebt, de ClickHouse® master, en de versie waarnaar je wilt updaten. Je test immers je analytics-queries.

Ten slotte: test niet op een enkele instance; test op een cluster met minstens twee machines en Zookeeper of ClickHouse® Keeper. Het gedrag is niet hetzelfde; het maken van tabellen kan makkelijk mislukken, je zult replicatie-lag ervaren en er zijn andere kleine details die je wilt testen.

Kosten

De rekensom voor het draaien van een cluster is eenvoudig:

• Kosten van de replica's/shards. Dit hangt natuurlijk af van de prijs van je infrastructuur.
• Minstens 3 ZooKeeper-replica's. Deze moeten op een andere machine/pod staan dan de database. Je wilt dat je ZK-server zo geïsoleerd mogelijk is. Als ZK op dezelfde machine staat en die machine raakt overbelast, wordt alles traag en uiteindelijk faalt het, waardoor je tabellen in read-only modus komen. Een ZK-probleem betekent dat je echt een probleem hebt.
• Opslag: Beslis waar je de data opslaat (lokaal, S3 of beide) en reken dit door. Houd er rekening mee dat schijven niet kunnen worden verkleind en dat je bij S3 rekening moet houden met de kosten voor operaties.
• Waarschijnlijk een loadbalancer. Een kleine is voldoende, maar je hebt HA (High Availability) nodig, dus een paar stuks.
• Backup-opslag: Afhankelijk van het retentiebeleid en de hoeveelheid data kan dit duur zijn.

Hoe bepaal je hoeveel cores je nodig hebt? Dit hangt sterk af van je belasting en use case. Als vuistregel: een machine met 32 cores kan gezond ongeveer 5GB/s (ongecomprimeerde data) max verwerken.

Over het personeel: Voor een klein cluster is één persoon parttime voldoende. Voor belastingen van meer dan 20k rows/s en mensen die constant wijzigingen doorvoeren, heb je wellicht een fulltime beheerder nodig om het cluster te beheren en een oogje in de wacht te houden bij de bizarre queries die mensen gaan schrijven. Er zit een factor 3-4 verschil in de benodigde hardware als je een paar basisregels volgt. Een expert in huis hebben verdient zich dus terug.

Wanneer je dat punt bereikt, is het inhuren van de juiste mensen net zo kritisch als het kiezen van de juiste hardware. Goede ClickHouse®- en data-infrastructure engineers zijn moeilijk te vinden.

Ik raad niet aan om een cluster volledig in je eentje te beheren. Je wilt je analytics-productvereisten oplossen, niet je tijd besteden aan het beheren van ClickHouse®. Daarom hebben we Tinybird gemaakt. Laat me duidelijk zijn: we bieden geen ClickHouse®-hosting aan; we lossen het analytics-probleem op. Dat is een smaller probleem dan wat een data warehouse oplost, en wij lossen het beter op.

Ingestie

Elk bedrijf dat ClickHouse® beheert, worstelt met ingestie. Ze verliezen data (vaak zonder het te weten), creëren duplicaten of brengen de database ten val.

Ingestie is niet alleen het uitvoeren van inserts; het is een fijngevoelig evenwicht tussen:

• Merges
• Inserts
• Reads
• Mutations
• Tabelontwerp

Het is een kunst om dit goed in te richten.

De basis van ClickHouse®: Wanneer je een insert uitvoert, wordt er een nieuwe 'part' gegenereerd. Van tijd tot tijd worden deze parts in een achtergrondproces samengevoegd (merged) tot een grotere part. Als je bekend bent met Iceberg, is dit het compactieproces.

Je wilt grote parts zodat reads sneller zijn (reads zijn sneller als ze data uit minder parts hoeven te halen). Maar als parts groot zijn, kosten merges veel tijd, CPU en geheugen. Je zult merken dat je cluster af en toe vastloopt door merges.

Je zou kunnen denken: "laten we de insert-batches groter maken zodat er minder parts worden gegenereerd". Dat klopt, maar grotere parts betekenen meer latentie omdat je moet wachten op meer data, en meer geheugengebruik bij inserts, wat kan leiden tot Out-Of-Memory (OOM) fouten.

De typische situatie is een query die alle CPU en/of IO verbruikt, waardoor inserts zich opstapelen, het geheugen groeit, wat leidt tot een OOM en dus tot dataverlies.

Mijn aanbevelingen voor ingestie:

• Batch inserts: Gebruik batches (er is async insert in ClickHouse®, maar die is beperkt). Niet te groot, niet te klein. Streef naar slechts 1 part per insert. Als je per partitie kunt batchen met verschillende flush-tijden, is dat beter.
• Frequentie: Voer frequente inserts alleen uit op tabellen die dat echt nodig hebben.
• Compact parts: Gebruik Compact parts als je S3 gebruikt (en zelfs als je dat niet doet). Dit bespaart veel write-operaties en voorkomt dat je de S3-rate limits bereikt.
• Hot disk: Soms bespaart een kleine "hot" schijf veel geld, omdat de writes en eerste merges daarop gebeuren en niet direct op S3.
• Max part size: Pas de max part size aan om extra merges te voorkomen, maar pas op dat ze niet té groot worden.
• Tabelpartitionering: Leer je team over partitionering. Je kunt tabelontwerp niet loskoppelen van ingestie. Een verkeerde partition key of een slecht ontworpen materialized view kan het cluster kapotmaken.

Geheugen is een groot probleem, vooral bij materialized views (MV's). Een MV die een klein beetje verkeerd is ontworpen en meer geheugen verbruikt dan nodig, veroorzaakt een OOM op de server. Wij hebben een systeem dat MV's in zulke gevallen ontkoppelt.

Andere problemen:

• Tabellen in read-only modus: Meestal omdat de replica aan het opstarten is. Wacht even en controleer voordat je writes verplaatst. Soms gebeurt dit om "onbekende redenen". Een snelle oplossing: drop de tabel, maak hem opnieuw aan en laat replicatie het werk doen.
• Too many parts: Gebeurt wanneer je te veel parts per write schrijft (bijv. partitioneren per dag en 3 jaar aan data in één keer inserten). Probeer data zo te schrijven dat het in één partitie landt.
• Merges zijn trager dan inserts: Gebeurt wanneer je te snel insert en de merge-queue te groot wordt. Verhoog de thread pool, pas backpressure toe op de ingestie of wijzig instellingen om de queue verder te laten groeien.

Overige zaken:

• Backpressure mechanisme: Sommigen plaatsen Kafka voor ClickHouse®. Dit werkt, maar kan duur zijn. Wij hebben een eigen systeem ontwikkeld omdat we multi-tenant zijn en de ingestie per tabel willen kunnen tweaken. In productie is zo'n mechanisme essentieel om dataverlies bij incidenten te voorkomen.
• Plotselinge pieken: Je hebt niet altijd tijd om een nieuwe replica op te spinnen. Je hebt een queue nodig die deze pieken opvangt.
• Gedupliceerde data: Als een insert faalt en je systeem is niet zorgvuldig ontworpen, eindig je met duplicaten, waardoor materialized views breken en statistieken onjuist worden.

Dit is het eerste deel. In deel 2 behandel ik: het beheren en monitoren van het cluster (overweeg Grafana), het omgaan met load, prestatie-tweaks en andere veelvoorkomende problemen.