> **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. # Loaders A loader writes a query's results into another system: a database like PostgreSQL, a warehouse like Snowflake, or a data lake. Use one to keep Amp's data next to your own, or to serve it with tools that don't speak to Amp. The SDK creates the destination table from the results' columns, and loads the rows in batches, without converting them through Python objects. ## 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 examples, the `deltalake` package, which the [Delta Lake loader](/sdks/python/loaders/delta-lake) uses: `pip install deltalake` or `uv add deltalake`. It writes to a folder on your disk, so it needs no server. ## Loaders | Destination | Loader name | Packages | Modes | | --------------------------------------------- | ------------ | -------------------------------------------------------------------------- | -------------------------------------------------------------------------------- | | [PostgreSQL](/sdks/python/loaders/postgresql) | `postgresql` | `psycopg2-binary` | `append`, `overwrite` | | [Redis](/sdks/python/loaders/redis) | `redis` | `redis` | `append`, `overwrite` | | [Snowflake](/sdks/python/loaders/snowflake) | `snowflake` | `snowflake-connector-python`, and more for some loading methods | `append` | | [BigQuery](/sdks/python/loaders/bigquery) | `bigquery` | `google-cloud-bigquery`, and `google-cloud-storage` for load jobs | `append` | | [ClickHouse](/sdks/python/loaders/clickhouse) | `clickhouse` | `clickhouse-connect` | `append` | | [Delta Lake](/sdks/python/loaders/delta-lake) | `deltalake` | `deltalake` | `append`, `overwrite` | | [Iceberg](/sdks/python/loaders/iceberg) | `iceberg` | `pyiceberg`, with the extra for your catalog, like `pyiceberg[sql-sqlite]` | `append`, `overwrite`, and `upsert` and `merge` on tables with identifier fields | | [LMDB](/sdks/python/loaders/lmdb) | `lmdb` | `lmdb` | `append`, `overwrite` | Each loader creates what it writes to, like a table, if it doesn't exist (except a PostgreSQL or ClickHouse stream, which [expects the table to be there](/sdks/python/streaming#postgresql-and-clickhouse-dont-create-the-table)), adds an `_amp_batch_id` column to the rows of a stream, and undoes [reorgs](/sdks/python/streaming#reorgs). Its page shows what it creates, and how it undoes a reorg. For a destination that isn't here, write a [custom loader](/sdks/python/loaders/custom). ### Install a loader's packages The `amp` package doesn't install any destination's client library. Install the ones the loader you use needs, from the table above, the way you [installed the SDK](/sdks/python#installation): :::code-group ```bash [pip] pip install deltalake ``` ```bash [uv] uv add deltalake ``` ::: `client.get_available_loaders()` lists the loaders the SDK can find, and what happens without a loader's package depends on the loader. PostgreSQL, Redis, LMDB, and Snowflake aren't listed, and using one fails with `Loader 'postgresql' not found. Available loaders: […]`. BigQuery, ClickHouse, Delta Lake, and Iceberg are listed, and fail when you create or connect them, with an `ImportError` or `ModuleNotFoundError` naming the package. ## Load query results A loader writes to a destination through a connection: the loader's name, and its configuration, like a host and a password, which each loader's page documents. `configure_connection()` names a connection, and `load()` uses it: ```python [load.py] from amp import Client client = Client('grpc://localhost:1602') client.configure_connection('lake', 'deltalake', {'table_path': './lake/transactions'}) result = client.sql('SELECT block_num, tx_index, tx_hash, gas_used FROM "acme/eth_mainnet".transactions').load( connection='lake', destination='transactions', read_all=True ) print(result) print(result.success, result.rows_loaded, result.metadata['batches_processed']) ``` :::terminal ```bash python load.py ``` ``` ✅ Loaded 17153 rows to transactions in 0.12s True 17153 3 ``` ::: * `connection` is the connection's name, and `destination` what the loader writes to. Each loader reads it its own way, like a table name or a key prefix. Delta Lake ignores it, and writes to its `table_path`. * `read_all=True` runs the query, holds all the results in memory, then loads them in the batches Amp sent, split into batches of `batch_size` rows when they're larger. * `load()` returns a [`LoadResult`](/sdks/python/reference/loaders#loadresult): whether it succeeded, how many rows it loaded, how long it took, and details in `metadata`. ### Load in batches Without `read_all=True`, `load()` loads each batch as Amp sends it, so it never holds more than one batch in memory. It returns an iterator, with a `LoadResult` per batch, which loads a batch each time your loop asks for the next result: ```python [load_batches.py] from amp import Client client = Client('grpc://localhost:1602') client.configure_connection('lake', 'deltalake', {'table_path': './lake/transactions'}) results = client.sql('SELECT block_num, tx_index, tx_hash, gas_used FROM "acme/eth_mainnet".transactions').load( connection='lake', destination='transactions' ) for result in results: print(result) ``` :::terminal ```bash python load_batches.py ``` ``` ✅ Loaded 8192 rows to transactions in 0.02s ✅ Loaded 8192 rows to transactions in 0.02s ✅ Loaded 769 rows to transactions in 0.02s ✅ Loaded 0 rows to transactions in 0.01s ``` ::: The table now has 34,306 rows: both scripts' rows, since loads [append](/sdks/python/loaders#load-modes) by default. :::warning Nothing loads until you loop over the results. A script that calls `load()` without `read_all=True`, and doesn't loop, loads nothing, without an error. ::: To load continuously, as new blocks arrive, pass `stream=True`: see [Streaming](/sdks/python/streaming). ### Check for failures `load()` doesn't raise an exception when the destination rejects the rows: it returns a `LoadResult` whose `success` is `False`, with the reason in `error`. Check it: ```python if not result.success: raise RuntimeError(f'Loading failed: {result.error}') ``` In a loop of results, a failed batch also stops the loop: the SDK doesn't skip rows, so the destination doesn't end up with a gap. ## Connections `configure_connection(name, loader, config)` names a connection, so the rest of your code only needs its name. :::warning The SDK saves every connection you configure, with its configuration, passwords included, in plain text, in `~/.flight_sql_connections.json`. Later scripts can use them by name without configuring them again. To keep passwords out of that file, configure connections in each script, with values from environment variables, and delete the file; or set the `FLIGHT_SQL_CONNECTIONS` environment variable to another path. ::: * `client.list_connections()` returns your saved connections' names and loaders, like `{'lake': 'deltalake'}`. * A connection name that isn't saved falls back to environment variables for PostgreSQL and Redis: a name that starts or ends with `postgresql` reads `DATABASE_URL`, `POSTGRESQL_URL`, or `POSTGRES_URL`, like `postgresql://postgres:postgres@localhost:5432/postgres`, and one that starts or ends with `redis` reads `REDIS_URL`. To skip naming a connection, pass its configuration to `load()` instead, with the loader's name under `loader_type`, and `connection=None`: ```python result = client.sql('SELECT block_num, gas_used FROM "acme/eth_mainnet".blocks').load( connection=None, destination='blocks', config={'loader_type': 'deltalake', 'table_path': './lake/blocks'}, read_all=True, ) ``` ## Load options `load()` takes these, besides `connection`, `destination`, and `config`: | Option | Type | Default | Description | | -------------- | ------------------------------------------------------------------- | -------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `read_all` | `bool` | `False` | Holds all the results in memory, then loads them, and returns one `LoadResult`. Otherwise, loads each batch as it arrives. | | `batch_size` | `int` | `10000` | With `read_all=True`, the most rows a batch the loader writes has: it splits larger batches from Amp, but doesn't combine smaller ones. Otherwise, Amp decides the batches' sizes. | | `mode` | `str`: `append`, `overwrite`, `upsert`, or `merge` | `append` | How the rows join the table's existing rows. See [Load modes](/sdks/python/loaders#load-modes). | | `create_table` | `bool` | `True` | Creates the table if it doesn't exist, from the results' columns. With `False`, the table must exist. | | `label_config` | [`LabelJoinConfig`](/sdks/python/reference/loaders#labeljoinconfig) | `None` | Adds columns from a label file to each row, and drops the rows without one. Ignored when streaming. See [Labels](/sdks/python/labels). | | `stream` | `bool` | `False` | Loads continuously, as new blocks arrive. See [Streaming](/sdks/python/streaming), which lists the options that go with it. | Other options go to the loader: each loader's page lists the ones it takes. `load()` also accepts `max_retries`, `retry_delay`, and `schema_evolution`, but loaders don't use them: set retries with [`resilience`](/sdks/python/loaders#retries), and schema changes in the loader's configuration, where it has one. ### Load modes | Mode | What happens to the table's existing rows | | ----------- | ------------------------------------------------------------------------------ | | `append` | They stay. The new rows are added. | | `overwrite` | They're deleted, then the new rows are added. | | `upsert` | Rows that match a new row on the table's key are replaced by it. Iceberg only. | | `merge` | The same as `upsert`. Iceberg only. | A loader that doesn't support a mode returns a failed result: `Unsupported mode LoadMode.UPSERT. Supported modes: …`. :::warning `overwrite` doesn't do what it says in every case: * **In batches**, without `read_all=True`, the loader deletes the existing rows before each batch, and Amp's last batch is empty, so the table ends up empty. Use `overwrite` only with `read_all=True`. * **PostgreSQL** never deletes the rows of a table that existed before the load. * **Delta Lake and Iceberg** replace the table's rows with each batch Amp sends, so only the last one stays, even with `read_all=True` and a large `batch_size`. * **LMDB** deletes the whole database, not only the destination's rows. The results don't show any of this: they count every row written, kept or not. To replace a table's rows reliably, delete them yourself, then load in `append` mode. Each loader's page has the details, and the [Delta Lake](/sdks/python/loaders/delta-lake#load-modes) and [Iceberg](/sdks/python/loaders/iceberg#load-modes) pages show how. ::: ## Retries When a batch fails to load because of a problem that may pass, like a timeout, a connection error, or a rate limit, the loader waits, then tries again, waiting longer after each failure. When the destination limits how fast it accepts data, the loader also slows down, and speeds up again as batches succeed. Any other error, like a missing column, fails the batch at once. PostgreSQL and ClickHouse streams don't retry or slow down, with state on (the default): a batch that fails returns a failed result at once. One-off loads into them do retry. Each loader takes these settings under a `resilience` key of its configuration: ```python client.configure_connection( 'lake', 'deltalake', {'table_path': './lake/transactions', 'resilience': {'retry': {'max_retries': 10}}}, ) ``` | Key | Type | Default | Description | | -------------------------------- | ------- | -------- | ------------------------------------------------------------------------ | | `retry.enabled` | `bool` | `True` | Retries failed batches. | | `retry.max_retries` | `int` | `5` | How many times to retry a batch before failing it. | | `retry.initial_backoff_ms` | `int` | `2000` | How long to wait before the first retry, in milliseconds. | | `retry.max_backoff_ms` | `int` | `120000` | The longest wait between retries, in milliseconds. | | `retry.backoff_multiplier` | `float` | `2.0` | How much longer each wait is than the one before. | | `retry.jitter` | `bool` | `True` | Varies each wait randomly, so many clients don't retry at the same time. | | `back_pressure.enabled` | `bool` | `True` | Slows down when the destination limits you or times out. | | `back_pressure.initial_delay_ms` | `int` | `0` | The delay before each batch at the start, in milliseconds. | | `back_pressure.max_delay_ms` | `int` | `5000` | The longest delay before a batch, in milliseconds. | | `back_pressure.adapt_on_429` | `bool` | `True` | Slows down when the destination answers that you send too much. | | `back_pressure.adapt_on_timeout` | `bool` | `True` | Slows down when the destination times out. | | `back_pressure.recovery_factor` | `float` | `0.9` | What the delay is multiplied by after each batch that succeeds. | The SDK decides that an error may pass when its message contains words like `timeout`, `429`, `503`, `504`, `connection reset`, `connection refused`, `rate limit`, or `throttle`. ## Troubleshooting ### `Loader 'postgresql' not found. Available loaders: […]` The loader's package isn't installed, or the name is wrong. Install the package from [the table](/sdks/python/loaders#loaders), then check the name. Through `load()`, the message comes back in the `LoadResult`'s `error`, rather than being raised. ### `Connection 'pg' not found. Available: ['lake']` No connection has that name. Configure it with `configure_connection()` first, in this script or an earlier one, or check its spelling. `client.list_connections()` lists the saved ones. ### `When using inline config, 'loader_type' must be specified` You passed `config` to `load()` without the loader's name. Add `'loader_type': 'deltalake'`, or the loader you use, to it. ### `'replace' is not a valid LoadMode` `mode` must be `append`, `overwrite`, `upsert`, or `merge`. ### `FATAL: Permanent error loading batch (not retryable)` The destination rejected a batch for a reason that won't pass by retrying, which the rest of the message gives. Fix it, then load again. The SDK stops rather than skip the batch. ### `FATAL: Max retries (5) exceeded for batch` A batch kept failing with errors that may pass, like timeouts. Check that the destination is reachable, or raise `resilience.retry.max_retries`.