> **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. # Streaming Status: Unstable. A [streaming query](/querying/streaming-queries) doesn't stop at the data that's there when it runs: Amp sends the rows there are, then new rows as it extracts new blocks, until you stop it. Use one to keep another system up to date with the chain, or to react to new blocks as they come. The Python SDK streams in two ways: * **Read the stream yourself**, batch by batch, with `stream()`. * **Load it continuously** into another system with a [loader](/sdks/python/loaders), with `load(stream=True)`. The SDK then also tracks which blocks each row came from, so it can undo rows from blocks that a [reorg](/get-started/concepts/reorgs) replaced. To load a long range of past blocks quickly first, see [Parallel Streaming](/sdks/python/parallel-streaming). :::warning The current SDK has two issues that affect every stream. Until they're fixed: * It ends every stream 30 seconds after it starts, even while new blocks keep arriving. To keep streaming, start the stream again in a loop, as [Streams end after 30 seconds](/sdks/python/streaming#streams-end-after-30-seconds) shows. * It reports a reorg at the end of every block range it loads, and then loads the same rows again, without end. Pass `with_reorg_detection=False` to `load()`, as the examples do. See [Known issues](/sdks/python/streaming#known-issues). ::: ## Prerequisites * The Python SDK, [installed](/sdks/python#installation). * The Amp and the `acme/eth_mainnet` dataset from the [Quickstart](/get-started/quickstart), running on your machine. * For the loading example, the `deltalake` package, which the [Delta Lake loader](/sdks/python/loaders/delta-lake) uses: `pip install deltalake` or `uv add deltalake`. The Quickstart's deployment stops at block 24,000,099, so a streaming query of its dataset sends those blocks, then has nothing more to send. On a dataset that follows the chain, the same examples also print new rows as Amp extracts new blocks. Either way, the SDK ends the stream 30 seconds after it starts (see [Streams end after 30 seconds](/sdks/python/streaming#streams-end-after-30-seconds)). ## Read a streaming query Add `SETTINGS stream = true` to the end of a query to make it a streaming query, then read it with `stream()`, which yields each Arrow record batch as it arrives: ```python [stream_query.py] from amp import Client client = Client('grpc://localhost:1602') query = client.sql('SELECT block_num, gas_used FROM "acme/eth_mainnet".blocks SETTINGS stream = true') for batch in query.stream(): if batch.num_rows: blocks = batch.column('block_num') print(f'{batch.num_rows} rows, blocks {blocks[0]} to {blocks[-1]}') print('The stream ended') ``` :::terminal ```bash python stream_query.py ``` ``` 100 rows, blocks 24000000 to 24000099 The stream ended ``` ::: `The stream ended` comes 30 seconds after the stream starts: Amp keeps it open, but the SDK ends it (see [Streams end after 30 seconds](/sdks/python/streaming#streams-end-after-30-seconds)). Amp also sends batches with no rows, which mark the end of a block range, so the loop skips them. `stream()` gives you the rows only, not the block ranges they came from, so it can't tell you about reorgs: when a reorg replaces blocks, Amp sends the new blocks' rows, and the rows you already read from the replaced blocks stay wrong. To handle reorgs, load the stream with a loader instead. ## Load a stream continuously `load()` with `stream=True` runs the query as a streaming query, adding `SETTINGS stream = true` if it isn't there, and loads each batch into the destination as it arrives. This example loads the blocks into a Delta Lake table, a folder of files on your disk: ```python [stream_load.py] from amp import Client client = Client('grpc://localhost:1602') client.configure_connection('lake', 'deltalake', {'table_path': './blocks'}) results = client.sql('SELECT block_num, hash, gas_used FROM "acme/eth_mainnet".blocks').load( connection='lake', destination='blocks', stream=True, with_reorg_detection=False ) for result in results: print(result) print(' ', result.metadata['block_ranges'], 'complete' if result.metadata['ranges_complete'] else '') ``` :::terminal ```bash python stream_load.py ``` ``` Failed to load checkpoint, starting from beginning: 'DeltaLakeLoader' object has no attribute 'checkpoint_store' ✅ Loaded 100 rows to blocks in 0.02s [{'network': 'mainnet', 'start': 24000000, 'end': 24000099}] ✅ Loaded 0 rows to blocks in 0.01s [{'network': 'mainnet', 'start': 24000000, 'end': 24000099}] complete ``` ::: * **Results.** `load()` returns an iterator, which loads a batch each time your loop asks for the next result. Nothing loads until you loop over it. Each result is a [`LoadResult`](/sdks/python/reference/loaders#loadresult): how many rows it loaded, and in its `metadata`, the block ranges the batch covers, and whether it completes them. * **Batches.** Amp decides the batches' sizes. It sends a range's rows in one or more batches, then an empty batch that marks the range complete, which the SDK uses to record that the range is loaded. * **The first line** is a warning the SDK always prints: it looks for a saved position to resume from, which doesn't exist yet (see [Known issues](/sdks/python/streaming#known-issues)). * **Stopping.** The loop ends when the SDK ends the stream, 30 seconds after it starts (see [Streams end after 30 seconds](/sdks/python/streaming#streams-end-after-30-seconds)). Ctrl+C ends it sooner, without an error (see [Ctrl+C ends a stream without an error](/sdks/python/streaming#ctrlc-ends-a-stream-without-an-error)). Each row the loader writes gets an extra column, `_amp_batch_id`, which identifies the block range it came from: ```python [read_lake.py] from deltalake import DeltaTable table = DeltaTable('./blocks').to_pyarrow_table() print(table.num_rows, table.column_names) ``` :::terminal ```bash python read_lake.py ``` ``` 100 ['block_num', 'hash', 'gas_used', '_amp_batch_id'] ``` ::: Each loader's page shows how it stores that column, and the shared options of `load()`, like `mode`, are on [Loaders](/sdks/python/loaders#load-options). ### Streaming options `load()` takes these when `stream=True`: | Option | Type | Default | Description | | ---------------------- | --------------------------------------------------------------------- | ------- | ------------------------------------------------------------------------------------------------------------------------------------------------------- | | `stream` | `bool` | `False` | Loads continuously, as described on this page. | | `with_reorg_detection` | `bool` | `True` | Detects reorgs, and undoes the rows of replaced blocks. See [Reorgs](/sdks/python/streaming#reorgs). | | `parallel_config` | [`ParallelConfig`](/sdks/python/reference/streaming#parallelconfig) | `None` | Loads past blocks in parallel first. See [Parallel Streaming](/sdks/python/parallel-streaming). | | `resume_watermark` | [`ResumeWatermark`](/sdks/python/reference/streaming#resumewatermark) | `None` | The block ranges you loaded before, with their hashes, so a reorg of those blocks is detected. Amp still streams from the start of the query's results. | `label_config`, which adds [labels](/sdks/python/labels) to rows, is ignored when streaming, except by the parallel workers of [parallel streaming](/sdks/python/parallel-streaming). ## Reorgs A reorg happens when the chain replaces its latest blocks with others: the rows Amp extracted from the replaced blocks are no longer part of the chain. [Reorgs](/get-started/concepts/reorgs) explains when and why. With each batch of a streaming query, Amp sends the block range it covers, on each network, with the hash of the range's last block and of the block before it. With `with_reorg_detection=True`, the SDK compares each range with the previous one on the same network, and detects a reorg when: * the new range starts at or before the end of the previous one, because Amp is sending those blocks again, or * the new range's previous block hash doesn't match the last block hash it received, because the block it builds on changed. When it detects one, the SDK: 1. Asks the loader to delete the rows of every batch it loaded from the start of the previous range on, using their `_amp_batch_id`. Each loader does it its own way: see its page. 2. Yields a `LoadResult` whose `is_reorg` is `True`, and whose `invalidation_ranges` lists the block ranges it undid. 3. Starts the stream again, to load the blocks that replaced them. ```python for result in results: if result.is_reorg: for block_range in result.invalidation_ranges: print(f'Reorg on {block_range.network}: undid blocks {block_range.start} to {block_range.end}') elif not result.success: print(f'Failed: {result.error}') ``` `_amp_batch_id` is a hash of the network, the range's first and last block numbers, the hash of its last block, and the hash of the block before it, so the same blocks from different forks of the chain get different IDs. A batch that covers several networks gets one ID per network, separated by `|`. ### Stream state To know which batches to delete, and which it already loaded, the SDK keeps a list of the batches it loaded: the stream state. By default, it keeps it in memory, so it lasts as long as your program runs. The [Snowflake loader](/sdks/python/loaders/snowflake) can keep it in Snowflake instead. Each loader takes these settings under a `state` key of its configuration: | Key | Type | Default | Description | | --------------------- | ------ | -------- | ----------------------------------------------------------------------------------------------------------------------------------------------------- | | `enabled` | `bool` | `True` | Tracks loaded batches. Without it, reorgs delete nothing, and duplicates aren't skipped. | | `storage` | `str` | `memory` | Where the state lives: `memory`, or `snowflake` for the Snowflake loader, which [fails](/sdks/python/loaders/snowflake#stream-state) at this version. | | `store_batch_id` | `bool` | `True` | Adds the `_amp_batch_id` column. Reorg handling needs it. | | `store_full_metadata` | `bool` | `False` | Also adds an `_amp_block_ranges` column, with each row's block ranges and hashes as JSON, for debugging. | ```python client.configure_connection( 'lake', 'deltalake', {'table_path': './blocks', 'state': {'store_full_metadata': True}}, ) ``` ## Known issues These affect the SDK at the commit these pages were written from. Each needs a fix in the SDK. ### Streams end after 30 seconds Every stream ends 30 seconds after it starts, without an error, even on a dataset that follows its chain, while new blocks keep arriving. So that the connection doesn't time out, Amp sends a message with no data every 30 seconds, counted from the start of the stream, and pyarrow, which the SDK reads streams with, takes that message for the end of the stream (see [Arrow Flight](/querying/arrow-flight#keep-alive-messages)). `stream()` and `load(stream=True)` both stop there, with every loader: on a Base dataset, which gets a new block every 2 seconds, both stopped at 30.1 seconds. The SDK doesn't resume a stream either (see [Restarting loads everything again](/sdks/python/streaming#restarting-loads-everything-again)), so start it again yourself, in a loop, from the block after the last one you have all the rows of. Filter on `_block_num`, a [reserved column](/datasets/manifests#reserved-columns): Amp then starts the stream at that block. With a filter on `block_num`, it reads the dataset's blocks from its first one again, to find no rows. To read a stream, start again after the last block you received: ```python [stream_forever.py] from amp import Client client = Client('grpc://localhost:1602') query = 'SELECT block_num, gas_used FROM "acme/eth_mainnet".blocks' last_block = None # The last block received while True: sql = query if last_block is None else f'{query} WHERE _block_num > {last_block}' for batch in client.sql(f'{sql} SETTINGS stream = true').stream(): if batch.num_rows: blocks = batch.column('block_num') print(f'{batch.num_rows} rows, blocks {blocks[0]} to {blocks[-1]}') last_block = blocks[-1].as_py() print(f'The stream ended. Starting it again after block {last_block}') ``` :::terminal ```bash python stream_forever.py ``` ``` 100 rows, blocks 24000000 to 24000099 The stream ended. Starting it again after block 24000099 The stream ended. Starting it again after block 24000099 ``` ::: The first stream sends the Quickstart's 100 blocks. The ones after it have no new blocks to send, and each ends 30 seconds after it starts, until you press Ctrl+C. `stream()` doesn't tell you where a block range ends, so this only works for a table with one row per block. In a table with several, like `logs`, a stream can end between two batches of the same block's rows, and starting again after that block skips the rest of its rows: load the stream instead. To load a stream, start again after the last block of the last complete range, which the loop saves in a file: ```python [stream_load_forever.py] from pathlib import Path from amp import Client from amp.streaming import StreamingResultIterator def stop(iterator, signum, frame): raise KeyboardInterrupt # Make Ctrl+C stop the program, rather than only end the stream StreamingResultIterator._handle_interrupt = stop client = Client('grpc://localhost:1602') client.configure_connection('lake', 'deltalake', {'table_path': './blocks'}) query = 'SELECT block_num, hash, gas_used FROM "acme/eth_mainnet".blocks' checkpoint = Path('last_block.txt') # The last block of the last complete range last_block = int(checkpoint.read_text()) if checkpoint.exists() else None while True: sql = query if last_block is None else f'{query} WHERE _block_num > {last_block}' results = client.sql(sql).load(connection='lake', destination='blocks', stream=True, with_reorg_detection=False) for result in results: print(result) if not result.success: raise RuntimeError(result.error) if result.metadata.get('ranges_complete'): last_block = result.metadata['block_ranges'][0]['end'] checkpoint.write_text(str(last_block)) print(f'The stream ended. Starting it again after block {last_block}') ``` :::terminal ```bash python stream_load_forever.py ``` ``` Failed to load checkpoint, starting from beginning: 'DeltaLakeLoader' object has no attribute 'checkpoint_store' ✅ Loaded 100 rows to blocks in 0.07s ✅ Loaded 0 rows to blocks in 0.01s The stream ended. Starting it again after block 24000099 Failed to load checkpoint, starting from beginning: 'DeltaLakeLoader' object has no attribute 'checkpoint_store' The stream ended. Starting it again after block 24000099 ``` ::: * **Ctrl+C.** Without `stop()`, which replaces the SDK's Ctrl+C handler, Ctrl+C only ends the stream, which the loop then starts again (see [Ctrl+C ends a stream without an error](/sdks/python/streaming#ctrlc-ends-a-stream-without-an-error)). * **Failed loads.** When a batch fails to load, the SDK yields a failed result. Most loaders then end the stream, but PostgreSQL and ClickHouse carry on with the next batches, including the empty one that completes the range. Saving that range would skip the rows that failed, so the loop stops at the first failed result, with a `RuntimeError`. Fix the cause, then run it again. * **Complete ranges only.** A stream can end partway through a range, after loading some of its rows. Starting after that range would skip the rest of its rows, so the loop starts after the last complete one, and loads the rows of the incomplete range again: they appear twice in the table. So do the rows of a range that loaded before a failed batch. * **Running it again** continues after the block in `last_block.txt`. Don't start after the last block in the table instead: in a table with several rows per block, like `logs`, that block's range can be incomplete. To load everything again, delete `last_block.txt` along with the table. * **Reorgs.** Amp only gets the block to start after, not its hash, so it can't tell when a reorg replaced blocks you already have. To send both, use the `amp-resume` header, which the SDK can't send (see [Resume a stream](/querying/streaming-queries#resume-a-stream)). ### Ctrl+C ends a stream without an error In your program's main thread, `load(stream=True)` handles Ctrl+C itself: it ends the stream, and your loop over the results ends as it does at the end of the stream, without a `KeyboardInterrupt`. Its handler then stays in place, so for the rest of your program, Ctrl+C does nothing. To make Ctrl+C raise `KeyboardInterrupt` instead, replace the SDK's handler before you stream, as [`stream_load_forever.py`](/sdks/python/streaming#streams-end-after-30-seconds) does: ```python from amp.streaming import StreamingResultIterator def stop(iterator, signum, frame): raise KeyboardInterrupt StreamingResultIterator._handle_interrupt = stop ``` `stream()` doesn't handle Ctrl+C, so Python raises `KeyboardInterrupt` as usual. Either way, Python only sees Ctrl+C when pyarrow receives Amp's next message: on a stream that's waiting for new blocks, that can be up to 30 seconds later. ### Every block range is reported as a reorg The empty batch that marks a block range complete has the same range as the batch before it, and its previous block hash is the hash of the block before the range, not of the range's last block. The SDK takes that mismatch for a reorg, at the end of every range: it logs `Reorg detected on mainnet: blocks 24000000-24000099 invalidated`, starts the stream again before recording the range as loaded, and loads the same rows again, without end. A range that Amp sends in more than one batch never completes, since the reorg is detected at its second batch. Pass `with_reorg_detection=False` to `load()`. The SDK then doesn't undo the rows of replaced blocks. To avoid reorgs altogether, stream from a dataset that only extracts finalized blocks, which no reorg can replace: generate its manifest with `ampctl manifest generate --finalized-blocks-only`. [Parallel streaming](/sdks/python/parallel-streaming#known-issues)'s hybrid mode always detects reorgs, so it has this issue too. ### Restarting loads everything again The SDK doesn't resume a stream where it stopped. When you run a streaming load again, Amp sends the query's results from the start, and the loader loads them again: running `stream_load.py` twice leaves 200 rows in the table. The warning `Failed to load checkpoint, starting from beginning` comes from the SDK looking for a saved position, which loaders don't have. Keeping the stream state in Snowflake doesn't prevent it: the SDK only checks whether it loaded a range at the empty batch that completes it, after loading its rows again. To restart without duplicates, start from a new or emptied table. To continue where you stopped, narrow the query to the blocks after the last range you loaded completely, like `WHERE _block_num > 24000099`, as [Streams end after 30 seconds](/sdks/python/streaming#streams-end-after-30-seconds) does. ### PostgreSQL and ClickHouse don't create the table In a streaming load, the [PostgreSQL](/sdks/python/loaders/postgresql) and [ClickHouse](/sdks/python/loaders/clickhouse) loaders write each batch in a transaction, which skips creating the table. A streaming load into a table that doesn't exist fails, with an error like `relation "blocks" does not exist`. Create the table first, with an empty load, as their pages show, or load with [parallel streaming](/sdks/python/parallel-streaming), which creates it.