ClickHouse sharding plan #8652

fuziontech · 2022-02-16T17:36:18Z

Plan for how we are going to shard ClickHouse on CH-Operator based installs and Cloud (and externally hosted CH as well)

macobo · 2022-02-22T09:22:57Z

There's multiple parts to this. Here's my thinking right now:

Problem being solved

Trying to scale vertically hits a limit for the users
Schemas locally/for our users and on cloud are out of sync

Also related: https://github.com/PostHog/product-internal/blob/main/requests-for-comments/2022-01-01-highscale-querying.md

Steps

1. Get schemas in sync

PostHog cloud and self-hosted have currently different schemas. CLICKHOUSE_REPLICATION env flag replicates some of this, but not all (and some is wrong, e.g. #7571).

Related PRs:

Make CLICKHOUSE_REPLICATION work correctly #8730

2. Support sharding in the helm chart

Code-wise should be relatively straight-forward as clickhouse-operator has options for this.

Some known subtasks:

Expose sharding options in values.yaml
Figure out what how to LB, how to set CLICKHOUSE_STABLE_HOST
ClickHouse service is created before all nodes are up - causing a race when creating the schema

Related PRs:

Set up sharding for clickhouse charts-clickhouse#310

3. Figure out how to upgrade from old schema to new

This is slightly tricky since table engines are changing - meaning we can't rename data-containing tables.

ATTACH PARTITION might work well for this without requiring any data swapping. See also https://stackoverflow.com/questions/68716267/is-it-possible-to-change-a-table-engine-of-an-existed-clickhouse-table

An annoyance here is the differentation between async migrations and normal clickhouse migrations. This will probably be an async migration even though it might not need to actually run anything long-running.

A related gotcha is posthog/async_migrations/migrations/0002_events_sample_by.py - this won't work due to no capability to run the code on all clickhouse nodes at this time. Will make it impossible to run that async migration if user is already sharded.

Related PRs:

Sharding: async migration to move to sharded schema #8783

4. Rebalancing data

We should figure out a mechanism for "rebalancing" data across a cluster. Note that this isn't strictly needed right now but operationally important for users scaling out.

Note that this logic can be relatively dumb for now and only work for self-hosted - e.g. we can stop ingestion if needed and even duplicate data temporarily as data is getting moved.

5. Documentation, removing `CLICKHOUSE_REPLICATION`

At this point we have a fully functioning story but it's not yet used anywhere.

Update documentation on how to set up sharding
Figure out if integrating with Altinity Cloud requires extra work here (Document using external ClickHouse cluster with Altinity Cloud charts-clickhouse#279)
Remove CLICKHOUSE_REPLICATION

Other information

Related issue: ClickHouse/ClickHouse#13574

yakkomajuri · 2022-02-22T14:29:38Z

As for 0002_events_sample_by won't we be above the limit version by the time sharding is introduced anyway?

macobo · 2022-02-22T14:33:07Z

As for 0002_events_sample_by won't we be above the limit version by the time sharding is introduced anyway?

See this:

An annoyance here is the differentation between async migrations and normal clickhouse migrations. This will probably be an async migration even though it might not need to actually run anything long-running.

So no, kind of not really due to the design of async migrations and that we need to set up new tables in clickhouse migrations.

macobo · 2022-03-01T13:29:50Z

Note: After investigating solutions for a while, I'll skip 4. Rebalancing data for now.

Rationale:

There's no great solutions right now built into alternative. The "best" building block right now is https://clickhouse.com/docs/en/sql-reference/statements/alter/partition/#alter_fetch-partition
No direct use-case right now. We can be reactive as users complain.
It will be affected by the solution to Future of our person model meta#39 as our sharding scheme changes

macobo · 2022-03-11T13:20:28Z

All the above has finished, some unexpected things have cropped up:

ATTACH PARTITION errors: Async migration 0004: Transaction failed (Node exists) #8920
Some of our core logic relies on CLICKHOUSE_REPLICATION to signal what the current schema is - e.g. materialization logic. Specifically fix(materialized columns): Create materialized columns without dollar sign in column name #8855 would break if we remove CLICKHOUSE_REPLICATION flag
- I'll update materialization logic to check events table engine instead
- cc note @yakkomajuri @tiina303 this is a case where there's an implicit dependency between async migration and clickhouse migration where one could be avoided if clickhouse migrations were also autorun async migrations

macobo · 2022-03-15T07:45:36Z

One last issue that needs resolving: clickhouse-operator only syncs some tables onto new nodes. See #8912 for the posthog-specific solution, but one table we don't replicate that way is the clickhouse migrations one.

I'll add a Distributed engine table to our fork of clickhouse orm.

macobo · 2022-03-18T08:52:04Z

Tested this by:

Setting up a new cluster with kind.
Deploying the following values.yaml:

cloud: local

Adding the make some noise plugin, to ingest some events.

┌─_table──────────────────────┬─count()─┐
│ events                      │   45765 │
│ person                      │    7599 │
│ person_distinct_id2         │    7595 │

Deployed the following values.yaml - includes all the sharding changes we've done for 1.34

cloud: "local"

image:
  repository: macobo/posthog-test
  sha: sha256:56d7b853ed4fffa158d869bd4877388252f96e35a628a3cba4407076d7a374a9

Ran the sharding async migration on the cluster
Verified schemas and event counts - all looks OK.
Deployed a sharding change:

cloud: "local"

image:
  repository: macobo/posthog-test
  sha: sha256:56d7b853ed4fffa158d869bd4877388252f96e35a628a3cba4407076d7a374a9

clickhouse:
  layout:
    shardsCount: 2
    replicasCount: 2

Verified the following tables on all shards:

Click to see SQL

SELECT
    hostName() AS host,
    name,
    engine_full
FROM clusterAllReplicas('posthog', system, tables)
WHERE name IN ('events', 'person', 'person_distinct_id', 'session_recording_events', 'sharded_events', 'sharded_session_recording_events', 'infi_clickhouse_orm_migrations', 'infi_clickhouse_orm_migrations_distributed')
ORDER BY
    name ASC,
    host ASC
FORMAT Vertical

Query id: a060b40a-4a2b-4793-87dc-0534b191662d

Row 1:
──────
host:        chi-posthog-posthog-0-0-0
name:        events
engine_full: Distributed('posthog', 'posthog', 'sharded_events', sipHash64(distinct_id))

Row 2:
──────
host:        chi-posthog-posthog-0-0-0
name:        events
engine_full: 

Row 3:
──────
host:        chi-posthog-posthog-0-1-0
name:        events
engine_full: Distributed('posthog', 'posthog', 'sharded_events', sipHash64(distinct_id))

Row 4:
──────
host:        chi-posthog-posthog-0-1-0
name:        events
engine_full: 

Row 5:
──────
host:        chi-posthog-posthog-1-0-0
name:        events
engine_full: Distributed('posthog', 'posthog', 'sharded_events', sipHash64(distinct_id))

Row 6:
──────
host:        chi-posthog-posthog-1-0-0
name:        events
engine_full: 

Row 7:
──────
host:        chi-posthog-posthog-1-1-0
name:        events
engine_full: Distributed('posthog', 'posthog', 'sharded_events', sipHash64(distinct_id))

Row 8:
──────
host:        chi-posthog-posthog-1-1-0
name:        events
engine_full: 

Row 9:
───────
host:        chi-posthog-posthog-0-0-0
name:        infi_clickhouse_orm_migrations
engine_full: MergeTree PARTITION BY toYYYYMM(applied) ORDER BY (package_name, module_name) SETTINGS index_granularity = 8192

Row 10:
───────
host:        chi-posthog-posthog-0-1-0
name:        infi_clickhouse_orm_migrations
engine_full: MergeTree PARTITION BY toYYYYMM(applied) ORDER BY (package_name, module_name) SETTINGS index_granularity = 8192

Row 11:
───────
host:        chi-posthog-posthog-1-0-0
name:        infi_clickhouse_orm_migrations
engine_full: MergeTree PARTITION BY toYYYYMM(applied) ORDER BY (package_name, module_name) SETTINGS index_granularity = 8192

Row 12:
───────
host:        chi-posthog-posthog-1-1-0
name:        infi_clickhouse_orm_migrations
engine_full: MergeTree PARTITION BY toYYYYMM(applied) ORDER BY (package_name, module_name) SETTINGS index_granularity = 8192

Row 13:
───────
host:        chi-posthog-posthog-0-0-0
name:        infi_clickhouse_orm_migrations_distributed
engine_full: Distributed('posthog', 'posthog', 'infi_clickhouse_orm_migrations', rand())

Row 14:
───────
host:        chi-posthog-posthog-0-1-0
name:        infi_clickhouse_orm_migrations_distributed
engine_full: Distributed('posthog', 'posthog', 'infi_clickhouse_orm_migrations', rand())

Row 15:
───────
host:        chi-posthog-posthog-1-0-0
name:        infi_clickhouse_orm_migrations_distributed
engine_full: Distributed('posthog', 'posthog', 'infi_clickhouse_orm_migrations', rand())

Row 16:
───────
host:        chi-posthog-posthog-1-1-0
name:        infi_clickhouse_orm_migrations_distributed
engine_full: Distributed('posthog', 'posthog', 'infi_clickhouse_orm_migrations', rand())

Row 17:
───────
host:        chi-posthog-posthog-0-0-0
name:        person
engine_full: ReplicatedReplacingMergeTree('/clickhouse/tables/am0004_20220318084324_noshard/posthog.person', '{replica}-{shard}', _timestamp) ORDER BY (team_id, id) SETTINGS index_granularity = 8192

Row 18:
───────
host:        chi-posthog-posthog-0-1-0
name:        person
engine_full: ReplicatedReplacingMergeTree('/clickhouse/tables/am0004_20220318084324_noshard/posthog.person', '{replica}-{shard}', _timestamp) ORDER BY (team_id, id) SETTINGS index_granularity = 8192

Row 19:
───────
host:        chi-posthog-posthog-1-0-0
name:        person
engine_full: ReplicatedReplacingMergeTree('/clickhouse/tables/am0004_20220318084324_noshard/posthog.person', '{replica}-{shard}', _timestamp) ORDER BY (team_id, id) SETTINGS index_granularity = 8192

Row 20:
───────
host:        chi-posthog-posthog-1-1-0
name:        person
engine_full: ReplicatedReplacingMergeTree('/clickhouse/tables/am0004_20220318084324_noshard/posthog.person', '{replica}-{shard}', _timestamp) ORDER BY (team_id, id) SETTINGS index_granularity = 8192

Row 21:
───────
host:        chi-posthog-posthog-0-0-0
name:        person_distinct_id
engine_full: CollapsingMergeTree(_sign) ORDER BY (team_id, distinct_id, person_id) SETTINGS index_granularity = 8192

Row 22:
───────
host:        chi-posthog-posthog-0-1-0
name:        person_distinct_id
engine_full: CollapsingMergeTree(_sign) ORDER BY (team_id, distinct_id, person_id) SETTINGS index_granularity = 8192

Row 23:
───────
host:        chi-posthog-posthog-1-0-0
name:        person_distinct_id
engine_full: CollapsingMergeTree(_sign) ORDER BY (team_id, distinct_id, person_id) SETTINGS index_granularity = 8192

Row 24:
───────
host:        chi-posthog-posthog-1-1-0
name:        person_distinct_id
engine_full: CollapsingMergeTree(_sign) ORDER BY (team_id, distinct_id, person_id) SETTINGS index_granularity = 8192

Row 25:
───────
host:        chi-posthog-posthog-0-0-0
name:        session_recording_events
engine_full: Distributed('posthog', 'posthog', 'sharded_session_recording_events', sipHash64(distinct_id))

Row 26:
───────
host:        chi-posthog-posthog-0-1-0
name:        session_recording_events
engine_full: Distributed('posthog', 'posthog', 'sharded_session_recording_events', sipHash64(distinct_id))

Row 27:
───────
host:        chi-posthog-posthog-1-0-0
name:        session_recording_events
engine_full: Distributed('posthog', 'posthog', 'sharded_session_recording_events', sipHash64(distinct_id))

Row 28:
───────
host:        chi-posthog-posthog-1-1-0
name:        session_recording_events
engine_full: Distributed('posthog', 'posthog', 'sharded_session_recording_events', sipHash64(distinct_id))

Row 29:
───────
host:        chi-posthog-posthog-0-0-0
name:        sharded_events
engine_full: ReplicatedReplacingMergeTree('/clickhouse/tables/am0004_20220318084324_{shard}/posthog.events', '{replica}', _timestamp) PARTITION BY toYYYYMM(timestamp) ORDER BY (team_id, toDate(timestamp), event, cityHash64(distinct_id), cityHash64(uuid)) SAMPLE BY cityHash64(distinct_id) SETTINGS index_granularity = 8192

Row 30:
───────
host:        chi-posthog-posthog-0-1-0
name:        sharded_events
engine_full: ReplicatedReplacingMergeTree('/clickhouse/tables/am0004_20220318084324_{shard}/posthog.events', '{replica}', _timestamp) PARTITION BY toYYYYMM(timestamp) ORDER BY (team_id, toDate(timestamp), event, cityHash64(distinct_id), cityHash64(uuid)) SAMPLE BY cityHash64(distinct_id) SETTINGS index_granularity = 8192

Row 31:
───────
host:        chi-posthog-posthog-1-0-0
name:        sharded_events
engine_full: ReplicatedReplacingMergeTree('/clickhouse/tables/am0004_20220318084324_{shard}/posthog.events', '{replica}', _timestamp) PARTITION BY toYYYYMM(timestamp) ORDER BY (team_id, toDate(timestamp), event, cityHash64(distinct_id), cityHash64(uuid)) SAMPLE BY cityHash64(distinct_id) SETTINGS index_granularity = 8192

Row 32:
───────
host:        chi-posthog-posthog-1-1-0
name:        sharded_events
engine_full: ReplicatedReplacingMergeTree('/clickhouse/tables/am0004_20220318084324_{shard}/posthog.events', '{replica}', _timestamp) PARTITION BY toYYYYMM(timestamp) ORDER BY (team_id, toDate(timestamp), event, cityHash64(distinct_id), cityHash64(uuid)) SAMPLE BY cityHash64(distinct_id) SETTINGS index_granularity = 8192

Row 33:
───────
host:        chi-posthog-posthog-0-0-0
name:        sharded_session_recording_events
engine_full: ReplicatedReplacingMergeTree('/clickhouse/tables/am0004_20220318084324_{shard}/posthog.session_recording_events', '{replica}', _timestamp) PARTITION BY toYYYYMMDD(timestamp) ORDER BY (team_id, toHour(timestamp), session_id, timestamp, uuid) TTL toDate(created_at) + toIntervalWeek(3) SETTINGS index_granularity = 512

Row 34:
───────
host:        chi-posthog-posthog-0-1-0
name:        sharded_session_recording_events
engine_full: ReplicatedReplacingMergeTree('/clickhouse/tables/am0004_20220318084324_{shard}/posthog.session_recording_events', '{replica}', _timestamp) PARTITION BY toYYYYMMDD(timestamp) ORDER BY (team_id, toHour(timestamp), session_id, timestamp, uuid) TTL toDate(created_at) + toIntervalWeek(3) SETTINGS index_granularity = 512

Row 35:
───────
host:        chi-posthog-posthog-1-0-0
name:        sharded_session_recording_events
engine_full: ReplicatedReplacingMergeTree('/clickhouse/tables/am0004_20220318084324_{shard}/posthog.session_recording_events', '{replica}', _timestamp) PARTITION BY toYYYYMMDD(timestamp) ORDER BY (team_id, toHour(timestamp), session_id, timestamp, uuid) TTL toDate(created_at) + toIntervalWeek(3) SETTINGS index_granularity = 512

Row 36:
───────
host:        chi-posthog-posthog-1-1-0
name:        sharded_session_recording_events
engine_full: ReplicatedReplacingMergeTree('/clickhouse/tables/am0004_20220318084324_{shard}/posthog.session_recording_events', '{replica}', _timestamp) PARTITION BY toYYYYMMDD(timestamp) ORDER BY (team_id, toHour(timestamp), session_id, timestamp, uuid) TTL toDate(created_at) + toIntervalWeek(3) SETTINGS index_granularity = 512

36 rows in set. Elapsed: 0.123 sec.

guidoiaquinti · 2022-05-16T13:46:51Z

👋 @macobo can we consider this ✅ ?

fuziontech added helm P0 Critical, breaking issue (page crash, missing functionality) labels Feb 16, 2022

fuziontech assigned macobo Feb 16, 2022

macobo added clickhouse enhancement New feature or request and removed P0 Critical, breaking issue (page crash, missing functionality) labels Feb 22, 2022

paolodamico added deployments Anything tied to self-hosted deployments of PostHog and removed clickhouse labels Feb 22, 2022

macobo mentioned this issue Mar 14, 2022

fix(sharding): Make sharding async migration more reliable #9012

Merged

macobo closed this as completed May 16, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

ClickHouse sharding plan #8652

ClickHouse sharding plan #8652

fuziontech commented Feb 16, 2022

macobo commented Feb 22, 2022 •

edited

Loading

yakkomajuri commented Feb 22, 2022

macobo commented Feb 22, 2022

macobo commented Mar 1, 2022

macobo commented Mar 11, 2022

macobo commented Mar 15, 2022

macobo commented Mar 18, 2022

guidoiaquinti commented May 16, 2022

ClickHouse sharding plan #8652

ClickHouse sharding plan #8652

Comments

fuziontech commented Feb 16, 2022

macobo commented Feb 22, 2022 • edited Loading

Problem being solved

Steps

1. Get schemas in sync

2. Support sharding in the helm chart

3. Figure out how to upgrade from old schema to new

4. Rebalancing data

5. Documentation, removing CLICKHOUSE_REPLICATION

Other information

yakkomajuri commented Feb 22, 2022

macobo commented Feb 22, 2022

macobo commented Mar 1, 2022

macobo commented Mar 11, 2022

macobo commented Mar 15, 2022

macobo commented Mar 18, 2022

guidoiaquinti commented May 16, 2022

macobo commented Feb 22, 2022 •

edited

Loading

5. Documentation, removing `CLICKHOUSE_REPLICATION`