> **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. # Parallel Streaming Status: Unstable. Loading a chain's history, millions of blocks, one batch after another takes a long time. Parallel streaming splits the block range into partitions, and loads several at once, each in its own worker. Use it for a backfill: the first load of a long range of past blocks. In hybrid mode, it then switches to an ordinary [stream](/sdks/python/streaming), to keep up with new blocks. ## Prerequisites * The Python SDK, [installed](/sdks/python#installation), with the PostgreSQL loader's package: `pip install psycopg2-binary` or `uv add psycopg2-binary`. * The Amp and the `acme/eth_mainnet` dataset from the [Quickstart](/get-started/quickstart), running on your machine. * A PostgreSQL database to load into, like the one the [PostgreSQL loader](/sdks/python/loaders/postgresql#prerequisites) starts in Docker, with its password in the `POSTGRES_PASSWORD` environment variable. ## Load a block range in parallel Pass a `ParallelConfig` to `load()`, with `stream=True`. This example loads the 100 blocks' transactions with four workers: ```python [parallel_load.py] import os from amp import Client from amp.streaming import ParallelConfig client = Client('grpc://localhost:1602') client.configure_connection( 'local_postgres', 'postgresql', { 'host': 'localhost', 'database': 'postgres', 'user': 'postgres', 'password': os.environ['POSTGRES_PASSWORD'], 'connection_params': {'options': '-c search_path=public'}, }, ) config = ParallelConfig( num_workers=4, table_name='"acme/eth_mainnet".transactions', min_block=24000000, max_block=24000100, ) results = client.sql('SELECT block_num, tx_index, tx_hash, gas_used FROM "acme/eth_mainnet".transactions').load( connection='local_postgres', destination='transactions', stream=True, parallel_config=config ) for result in results: partition = result.metadata['partition_metadata'] print(f'{result} (blocks {partition["start_block"]} to {partition["end_block"] - 1})') ``` ```bash python parallel_load.py ``` The workers report their progress as they go, on standard error, in lines that start with 🚀, 📦, and ✅ Worker. The loop gets one [`LoadResult`](/sdks/python/reference/loaders#loadresult) per partition, as each finishes, in any order. The table now has the 17,153 transactions of the 100 blocks. :::warning To load into Delta Lake, Iceberg, or LMDB, set `num_workers=1`. Their loaders can't share a destination between workers yet, whatever your query. With several workers: * [Delta Lake](/sdks/python/loaders/delta-lake#parallel-loads) repeats rows, even when every partition succeeds. * [Iceberg](/sdks/python/loaders/iceberg#parallel-loads) fails most partitions, on a new table. * [LMDB](/sdks/python/loaders/lmdb#parallel-loads) fails most partitions. The rows of a failed partition are missing. The loop gets a failed result for it, not an exception. ::: ## How it works 1. **Partitions.** The SDK splits the range from `min_block` up to, but not including, `max_block` into partitions. By default, it makes one per worker, of equal size. With `partition_size`, it makes partitions of that many blocks instead, and the workers take them in turn: 100 blocks with `partition_size=30` make partitions of 30, 30, 30, and 10 blocks. 2. **The table.** Before the workers start, the SDK runs your query with `LIMIT 1` to learn its columns, and creates the destination table, so the workers don't race to create it. This fails for a query with `GROUP BY`, `ORDER BY`, or `LIMIT`: see [Write a query that partitions](/sdks/python/parallel-streaming#write-a-query-that-partitions). Delta Lake's loader doesn't create its table here, but at the first write. 3. **Queries.** Each worker runs your query for its partition, as an ordinary query rather than a streaming one: the SDK adds `block_num >= 24000000 AND block_num < 24000025` to its `WHERE` clause, or adds a `WHERE` clause, and removes `SETTINGS stream = true`. It then loads the results in batches, through its own connection to the destination. 4. **Batch IDs.** Ordinary queries don't come with block hashes, so each partition's rows get an `_amp_batch_id` made from `table_name` and the partition's first and last blocks. The workers are threads of your Python program. They spend most of their time waiting for Amp and for the destination, so threads speed them up, but CPU-heavy work in your own code doesn't run in parallel. ### Write a query that partitions The SDK adds the block filter to your query's text, so the query must: * Select from one table that has the block column, `block_num` by default, under that name. For a table whose block column has another name, like `_block_num`, set `block_column`. * Have its `WHERE`, `GROUP BY`, `ORDER BY`, `LIMIT`, and `SETTINGS` clauses, if any, at the end of the outer query. :::warning Each partition runs your query on its own blocks, and the SDK never merges the results, so anything computed across rows only sees one partition. With any loader, the load succeeds and the values are wrong. With four workers, on the Quickstart's 100 blocks: * `SELECT COUNT(*) AS n` loads 4 rows, with `n` from 3,614 to 4,665, instead of 1 row with 17,153. * `SELECT MAX(block_num) AS m` loads 4 rows, one with the last block of each partition, instead of 1 row. * `SELECT "to", COUNT(*) AS n … GROUP BY "to"` and `SELECT DISTINCT "to"` load 6,933 rows instead of 5,506. * `ROW_NUMBER() OVER (ORDER BY block_num, tx_index)` starts again at 1 in each partition, and stops at 4,665 instead of 17,153. Other window functions see one partition too. * `LIMIT 10` loads 40 rows, and `ORDER BY gas_used DESC LIMIT 5` loads 20: the top 5 of each partition. Grouping by the block column is safe: `GROUP BY block_num` loads the right 100 rows, since all of a block's rows are in one partition. For anything else, load the whole range in one call, with `num_workers=1` and no `partition_size`, which makes a single partition, or compute it in the destination after you load. ::: :::warning Put a space, not just a line break, on each side of those keywords, or the SDK misses them: the query fails with a SQL error, or with `SETTINGS stream = true`, loads the whole table in every partition. ::: :::warning With `GROUP BY`, `ORDER BY`, or `LIMIT`, several workers can't load into a new table. The SDK puts its `LIMIT 1` where it makes the query invalid, so it doesn't create the table, and the workers race to create it. With PostgreSQL, most partitions fail, and their rows are missing. The loop gets a failed result for each, not an exception. Until it's fixed, create the table first: load one block that your query returns rows for with `num_workers=1`, then load the other blocks with as many workers as [your loader allows](/sdks/python/parallel-streaming#load-a-block-range-in-parallel). This splits your query too, so use it only for a query that stays correct when split, like one with `GROUP BY block_num`, or `ORDER BY` without `LIMIT`. For other aggregates, window functions, `DISTINCT`, and `LIMIT`, load the whole range in one call with one worker, or compute them in the destination, as the first warning of this section says. ::: A query that joins tables, or uses `UNION`, may get the filter in the wrong place. Check the rows you get on a short range first. In a query with no `WHERE` clause, a window with `PARTITION BY` and `ORDER BY` gets it wrong too: the SDK adds its filter before that `ORDER BY`, and the query fails with a SQL parse error. ## Hybrid mode: catch up, then stream Leave out `max_block`, and parallel streaming catches up to the chain's head, then keeps going: ```python config = ParallelConfig( num_workers=4, table_name='"acme/eth_mainnet".transactions', min_block=24000000, ) ``` 1. The SDK finds the last block with data, with `SELECT MAX(block_num) FROM` the `table_name`. `table_name` must be the table's name as SQL needs it, with its dataset. 2. The workers load the blocks from `min_block` up to that block. 3. The SDK starts a [stream](/sdks/python/streaming) of your query from `reorg_buffer` blocks before that block (200 by default), with reorg detection. The overlap catches a reorg of the last blocks the workers loaded. The stream loads those blocks a second time, so in `append` mode, their rows appear twice. The loop gets one result per partition, then one per batch of the stream, until you stop it. :::warning The stream that hybrid mode starts always detects reorgs, so it has the issue [every block range is reported as a reorg](/sdks/python/streaming#every-block-range-is-reported-as-a-reorg): it loads the same rows again and again. Until it's fixed, use a `max_block`, then start a stream yourself, from the next block, with `with_reorg_detection=False`. That stream ends 30 seconds after it starts, so start it in a loop, as [Streams end after 30 seconds](/sdks/python/streaming#streams-end-after-30-seconds) shows. ::: ## Resume an interrupted backfill If a backfill stops partway, some partitions finished and others didn't. The SDK doesn't resume it for you: it loads the whole range again. Resume it yourself: 1. Find the first block of the earliest partition that didn't finish: the one whose `✅ Worker … COMPLETE` line is missing. 2. Delete the rows from that block on, which unfinished partitions may have partly loaded. 3. Run the backfill again, with `min_block` set to that block. ## ParallelConfig `from amp.streaming import ParallelConfig`. | Field | Type | Default | Description | | ---------------------- | ------------------------------------------------------------------------------------- | ----------- | ----------------------------------------------------------------------------------------------------------------------------------- | | `num_workers` | `int`, 1 or more | Required | How many partitions load at once. | | `table_name` | `str` | Required | The table to partition, as SQL names it, like `'"acme/eth_mainnet".transactions'`. Hybrid mode queries its last block. | | `min_block` | `int` | `0` | The first block to load. | | `max_block` | `int` or `None` | `None` | The block to stop before. `None` uses hybrid mode. | | `partition_size` | `int`, 1 or more, or `None` | `None` | How many blocks each partition has. `None` makes one partition per worker. | | `block_column` | `str` | `block_num` | The column with the block number, which partitions filter on. | | `stop_on_error` | `bool` | `False` | Stops every worker, and raises the error, when a partition fails. Otherwise, the loop gets a failed result and the others carry on. | | `reorg_buffer` | `int` | `200` | In hybrid mode, how many blocks before the last loaded one the stream starts. | | `retry_config` | [`RetryConfig`](/sdks/python/reference/streaming#retryconfig) or `None` | `None` | How each worker retries a batch that fails to load. `None` uses the loader's. | | `back_pressure_config` | [`BackPressureConfig`](/sdks/python/reference/streaming#backpressureconfig) or `None` | `None` | How each worker slows down when the destination limits it. `None` uses the loader's. | `batch_size` isn't part of it: pass it to `load()`, like the other [load options](/sdks/python/loaders#load-options). ### Choose the number of workers Each worker holds one batch in memory at a time, and opens its own connection to the destination. More workers load faster until Amp, your network, or the destination can't keep up. Start with 4, and raise it while the destination keeps up, except with Delta Lake, Iceberg, and LMDB, which [need 1](/sdks/python/parallel-streaming#load-a-block-range-in-parallel). Many small partitions, with `partition_size`, balance the work better than one per worker when some blocks have much more data than others. ## Known issues * **Hybrid mode loads the same rows without end**, as described in [Hybrid mode](/sdks/python/parallel-streaming#hybrid-mode-catch-up-then-stream). * **Delta Lake, Iceberg, and LMDB need one worker**, as described in [Load a block range in parallel](/sdks/python/parallel-streaming#load-a-block-range-in-parallel). * **Resuming from saved state doesn't work.** The SDK has code to skip the blocks a backfill already loaded, when the loader keeps its [stream state](/sdks/python/streaming#stream-state) outside your program, which only Snowflake can do. The workers record their partitions under another connection name than the one the SDK looks up, so it never finds them, and loads the whole range again. * **The progress lines** always print on standard error. To hide them, redirect it: `python parallel_load.py 2> /dev/null` hides the SDK's warnings too. ## Troubleshooting ### `duplicate key value violates unique constraint "pg_type_typname_nsp_index"` Several workers tried to create the same new PostgreSQL table, because [your query has `GROUP BY`, `ORDER BY`, or `LIMIT`](/sdks/python/parallel-streaming#write-a-query-that-partitions). The partitions that succeeded are already in the table, and loading the range again adds their rows a second time. Drop the table, then load the range as that section says. ### `ValueError: parallel_config requires stream=True` Pass `stream=True` to `load()` along with `parallel_config`. ### `ValueError: min_block (10) must be < max_block (5)` `max_block` is the block to stop before, so it must be greater than `min_block`. To load a single block, `12`, use `min_block=12, max_block=13`. ### `Failed to detect max block` In hybrid mode, the SDK couldn't query the last block of `table_name`. Check that `table_name` is the table's full name in SQL, with the dataset in double quotes, and that the table has data.