> **Can't find what you're looking for?** Use `search_docs` on the docs MCP server at `https://docs.ampdb.io/api/mcp` to find what you need. # ClickHouse Loader The ClickHouse loader writes query results into a ClickHouse table. ClickHouse is a columnar database built for analytical queries, like aggregating millions of transfers by day. Choose it for dashboards and analytics over large amounts of chain data. The loader creates the table for you, and inserts rows in Arrow format, without converting them row by row. ## Prerequisites * **The Python SDK**, [installed](/sdks/python#installation), and the ClickHouse client it uses, `clickhouse-connect`: :::code-group ```bash [pip] pip install clickhouse-connect ``` ```bash [uv] uv add clickhouse-connect ``` ::: * **A running Amp with the Quickstart's dataset**, `acme/eth_mainnet`. [Quickstart](/get-started/quickstart) sets it up. * **A ClickHouse server you can write to**, through its HTTP interface. To run one locally in Docker, set a password for its `default` user, then start ClickHouse with it: ```bash export CLICKHOUSE_PASSWORD=clickhouse docker run --name amp-loader-clickhouse -e CLICKHOUSE_PASSWORD -p 8123:8123 -d clickhouse/clickhouse-server ``` ## Load data This script names a connection to your ClickHouse server, `local_clickhouse`, then loads the Quickstart's 100 blocks into a `blocks` table: ```python [load_clickhouse.py] import os from amp import Client client = Client('grpc://localhost:1602') client.configure_connection( 'local_clickhouse', 'clickhouse', {'host': 'localhost', 'password': os.environ['CLICKHOUSE_PASSWORD']}, ) query = """ SELECT block_num, arrow_cast(timestamp, 'Timestamp(Second, Some("UTC"))') AS timestamp, hash, gas_used FROM "acme/eth_mainnet".blocks """ result = client.sql(query).load(connection='local_clickhouse', destination='blocks', read_all=True) print(result) ``` ```bash python load_clickhouse.py ``` * `destination` is the table's name, in the database the connection uses: `default`, here. * `arrow_cast(timestamp, …)` gives the timestamps the time zone `UTC`. Amp's timestamps have the time zone `+00:00`, which ClickHouse doesn't accept: without the cast, the load fails (see [Troubleshooting](/sdks/python/loaders/clickhouse#time-zone-0000-is-not-supported)). Cast every timestamp column you load. * `read_all=True` loads the whole result at once and returns one `LoadResult`. [Loaders](/sdks/python/loaders) covers the other ways to load, and the options `load()` takes. `configure_connection()` saves the connection, password included, so later scripts can use `local_clickhouse` without configuring it again. [Loaders](/sdks/python/loaders) explains where it saves it. Query the table: :::terminal ```bash docker exec amp-loader-clickhouse clickhouse-client --password "$CLICKHOUSE_PASSWORD" -q 'SELECT block_num, timestamp, gas_used FROM blocks ORDER BY block_num DESC LIMIT 3' ``` ``` 24000099 2025-12-13 00:25:59.000000 10751338 24000098 2025-12-13 00:25:47.000000 36233422 24000097 2025-12-13 00:25:35.000000 52877862 ``` ::: ## The table it creates When the table doesn't exist, the loader creates it with a column for each column of your query, then an `_amp_batch_id` column, with a bloom filter index on it. `_amp_batch_id` identifies the block range each row came from, so the loader can delete the rows of a reorged range. It stays empty when you load all at once, like above, and is filled when you [stream](/sdks/python/loaders/clickhouse#streaming-and-reorgs). :::terminal ```bash docker exec amp-loader-clickhouse clickhouse-client --password "$CLICKHOUSE_PASSWORD" -q 'SHOW CREATE TABLE blocks FORMAT TSVRaw' ``` ``` CREATE TABLE default.blocks ( `block_num` UInt64, `timestamp` DateTime64(6, 'UTC'), `hash` FixedString(32), `gas_used` UInt64, `_amp_batch_id` Nullable(String), INDEX idx_amp_batch_id (_amp_batch_id) TYPE bloom_filter GRANULARITY 1 ) ENGINE = MergeTree ORDER BY tuple() SETTINGS index_granularity = 8192 ``` ::: The table uses the `MergeTree` engine, without a sorting key (`ORDER BY tuple()`). For large tables you query often by a column, like `block_num`, create the table yourself with that sorting key before loading: the loader uses an existing table as it is. It then needs an `_amp_batch_id Nullable(String)` column, for streaming. Each column gets the ClickHouse type that holds its values: | Query column type | ClickHouse type | | ---------------------------------- | ------------------------------------------- | | Signed and unsigned integers | `Int8` to `Int64`, `UInt8` to `UInt64` | | Floats | `Float32` and `Float64` | | Decimals | `Decimal` with the same precision and scale | | Booleans | `Bool` | | Strings, structs, and times of day | `String` | | Binary | `String` | | Fixed-size binary, like hashes | `FixedString` of the same size | | Timestamps | `DateTime64(6)`, with the time zone if any | | Dates | `Date` | | Lists and maps | `Array` and `Map` of the types above | | Anything else | `String` | Columns that can be null are wrapped in `Nullable`. ## Configuration The third argument of `configure_connection()`: | Key | Type | Default | Description | | ---------------------- | ------- | ------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | `host` | string | `'localhost'` | The server's host name or IP address. | | `port` | integer | `8123` | The port of the server's HTTP interface, like `8443` for HTTPS. | | `database` | string | `'default'` | The database to load into, like `'analytics'`. It must exist. | | `username` | string | `'default'` | The user to connect as. It needs to create tables, and to insert into and delete from them. | | `password` | string | `''` | The user's password. Read it from an environment variable, like `os.environ['CLICKHOUSE_PASSWORD']`. | | `send_receive_timeout` | integer | `300` | How many seconds to wait for the server to answer, like `600`. | | `query_limit` | integer | `0` | The most rows a query the loader runs may return. `0` means no limit. | | `compress` | boolean | `True` | Whether to compress the data sent to the server. | | `connection_params` | dict | `{}` | Other arguments for [clickhouse-connect](https://clickhouse.com/docs/integrations/python)'s `get_client()`, passed as they are, like `{'secure': True}` to connect over HTTPS. | | `batch_size` | integer | `100000` | Not used. Pass `batch_size` to `load()` instead (see [Loaders](/sdks/python/loaders)). | | `insert_block_size` | integer | `1048576` | Not used. | The `state` and `resilience` keys, which every loader takes, are on [Loaders](/sdks/python/loaders). `destination` must be a plain name, of letters, digits, `_`, and `$`, not starting with a digit. Prefix it with a database to load into another one, like `'analytics.blocks'`. ## Load modes The ClickHouse loader only supports `append`, the default: it inserts the rows, keeping the table's existing rows. Other modes fail with `Unsupported mode LoadMode.OVERWRITE`. To replace a table's rows, empty it yourself, then load: ```bash docker exec amp-loader-clickhouse clickhouse-client --password "$CLICKHOUSE_PASSWORD" -q 'TRUNCATE TABLE blocks' ``` ## Streaming and reorgs (Unstable) With `stream=True`, `load()` runs a [streaming query](/querying/streaming-queries) and keeps loading new rows as Amp extracts new blocks. [Streaming](/sdks/python/streaming) explains how it works. A stream doesn't create its table: create it first, with an empty load that isn't `read_all`, as this script does. Without it, every batch fails with `Table default.blocks_live does not exist`. ```python [stream_clickhouse.py] from amp import Client client = Client('grpc://localhost:1602') query = """ SELECT block_num, arrow_cast(timestamp, 'Timestamp(Second, Some("UTC"))') AS timestamp, hash, gas_used FROM "acme/eth_mainnet".blocks """ # A stream doesn't create its table, so create it with an empty load first for _ in client.sql(f'{query} LIMIT 0').load(connection='local_clickhouse', destination='blocks_live'): pass results = client.sql(query).load( connection='local_clickhouse', destination='blocks_live', stream=True, with_reorg_detection=False ) for result in results: print(result) ``` :::terminal ```bash python stream_clickhouse.py ``` ``` Failed to load checkpoint, starting from beginning: 'ClickHouseLoader' object has no attribute 'checkpoint_store' ✅ Loaded 100 rows to blocks_live in 0.06s ✅ Loaded 0 rows to blocks_live in 0.06s ``` ::: The first line is a warning every stream prints, and `with_reorg_detection=False` works around a problem with reorg detection: both are explained in [Known issues](/sdks/python/streaming#known-issues). The stream ends 30 seconds after it starts, even on a dataset that follows its chain: Amp sends a message every 30 seconds to keep the connection open, and pyarrow, which the SDK reads streams with, takes it for the end of the stream (see [Arrow Flight](/querying/arrow-flight#keep-alive-messages)). To keep loading, start the stream again in a loop, as [Streams end after 30 seconds](/sdks/python/streaming#streams-end-after-30-seconds) shows. Streamed rows have their `_amp_batch_id` set. When Amp reports a [reorg](/get-started/concepts/reorgs), the loader looks up the batches it loaded from the reorged blocks onward, and deletes their rows with a mutation: ```sql ALTER TABLE blocks_live DELETE WHERE `_amp_batch_id` IN ('7dce8312375e3370', …) SETTINGS mutations_sync = 1 ``` The bloom filter index lets ClickHouse skip the parts of the table without those batches. `mutations_sync = 1` makes the loader wait for the delete to finish before it loads more rows. The loader remembers which batches it loaded in memory, for as long as the `load()` call runs. When a stream restarts, it starts from the beginning of the query, and can't undo reorgs of the batches it loaded before. ## Troubleshooting ### `Time zone +00:00 is not supported` Your query returns a timestamp with Amp's `+00:00` time zone, which ClickHouse rejects. Cast the column to UTC, like `arrow_cast(timestamp, 'Timestamp(Second, Some("UTC"))') AS timestamp`. ### `Unsupported mode LoadMode.OVERWRITE` The ClickHouse loader only supports the `append` mode (see [Load modes](/sdks/python/loaders/clickhouse#load-modes)). ### `Table default.blocks_live does not exist` You're streaming into a table that doesn't exist, or passed `create_table=False` to `load()`. Create the table first (see [Streaming and reorgs](/sdks/python/loaders/clickhouse#streaming-and-reorgs)). ### `Invalid SQL table name: 'eth-blocks'` `destination` must be a plain name: letters, digits, `_`, and `$`, like `'eth_blocks'`. ### `default: Authentication failed: password is incorrect, or there is no user with such name` The password doesn't match the user's. Check `CLICKHOUSE_PASSWORD` in the terminal you run the script from.