> **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.

# Delta Lake Loader

The Delta Lake loader writes query results into a [Delta Lake](https://delta.io) table: Parquet files plus a log of
every change, in a directory on your disk or in cloud storage (Amazon S3, Azure, or Google Cloud Storage). Each write
creates a new version of the table, and readers can query older versions. Choose it to keep chain data in a data lake,
for engines like Spark, Databricks, DuckDB, or Polars to read. It needs no server.

## Prerequisites

* **The Python SDK**, [installed](/sdks/python#installation), and the Delta Lake library it uses, `deltalake`:

  :::code-group
  ```bash [pip]
  pip install deltalake
  ```

  ```bash [uv]
  uv add deltalake
  ```
  :::

* **A running Amp with the Quickstart's dataset**, `acme/eth_mainnet`. [Quickstart](/get-started/quickstart) sets it up.

## Load data

This script names a connection to a Delta table in `./data/blocks`, `local_delta`, then loads the Quickstart's 100
blocks into it:

```python [load_delta_lake.py]
from amp import Client

client = Client('grpc://localhost:1602')

client.configure_connection('local_delta', 'deltalake', {'table_path': './data/blocks'})

result = client.sql(
    'SELECT block_num, timestamp, hash, gas_used FROM "acme/eth_mainnet".blocks'
).load(connection='local_delta', destination='blocks', read_all=True)

print(result)
```

```bash
python load_delta_lake.py
```

The loader's name is `deltalake`, without an underscore.

`table_path` decides where the data goes, not `destination`: the loader creates the table there on its first write, and
`destination` only names the load, in its `LoadResult` and logs. To load into several tables, configure a connection for
each. A relative `table_path` is relative to the directory you run the script from.

`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.

Read the table back, with Polars, which the SDK installs:

```python [read_delta_lake.py]
import polars as pl

df = pl.read_delta('./data/blocks')
print(df.select('block_num', 'timestamp', 'gas_used').sort('block_num', descending=True).head(3))
```

:::terminal
```bash
python read_delta_lake.py
```

```
shape: (3, 3)
┌───────────┬─────────────────────────┬──────────┐
│ block_num ┆ timestamp               ┆ gas_used │
│ ---       ┆ ---                     ┆ ---      │
│ i64       ┆ datetime[μs, UTC]       ┆ i64      │
╞═══════════╪═════════════════════════╪══════════╡
│ 24000099  ┆ 2025-12-13 00:25:59 UTC ┆ 10751338 │
│ 24000098  ┆ 2025-12-13 00:25:47 UTC ┆ 36233422 │
│ 24000097  ┆ 2025-12-13 00:25:35 UTC ┆ 52877862 │
└───────────┴─────────────────────────┴──────────┘
```
:::

Delta Lake has no unsigned integers, so `block_num` and `gas_used`, unsigned 64-bit integers in Amp, are stored as
signed 64-bit integers, `i64`. Timestamps are stored to the microsecond.

## Cloud storage

Set `table_path` to a URL to write to cloud storage, and pass the credentials in `storage_options`, which the loader
gives to Delta Lake as they are. For example, for Amazon S3, with your credentials in the `AWS_ACCESS_KEY_ID` and
`AWS_SECRET_ACCESS_KEY` environment variables:

```python
client.configure_connection(
    's3_delta',
    'deltalake',
    {
        'table_path': 's3://my-bucket/amp/blocks',
        'storage_options': {'AWS_REGION': 'us-east-1'},
    },
)
```

| Storage              | `table_path`                                      | Credentials                                                      |
| -------------------- | ------------------------------------------------- | ---------------------------------------------------------------- |
| Your disk            | A path, like `./data/blocks`, or a `file://` URL  | None                                                             |
| Amazon S3            | `s3://bucket/path`                                | `AWS_ACCESS_KEY_ID` and `AWS_SECRET_ACCESS_KEY`                  |
| Azure Storage        | `az://container/path`, `abfs://…`, or `abfss://…` | `AZURE_STORAGE_ACCOUNT_NAME` and `AZURE_STORAGE_ACCOUNT_KEY`     |
| Google Cloud Storage | `gs://bucket/path` or `gcs://bucket/path`         | `GOOGLE_SERVICE_ACCOUNT`, the path to a service account key file |

The loader looks for the credentials in `storage_options`, then in environment variables of the same names, and logs a
warning when they're missing. [Delta Lake's storage
documentation](https://delta-io.github.io/delta-rs/usage/loading-table/) lists every option. On S3, the loader sets
`AWS_S3_ALLOW_UNSAFE_RENAME` to `'true'` unless you set it: S3 can't rename files atomically, so two programs must never
write to the same table at once.

## Configuration

The third argument of `configure_connection()`:

| Key                    | Type            | Default  | Description                                                                                                                                                                                       |
| ---------------------- | --------------- | -------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `table_path`           | string          | required | Where the table is, as a path or a URL (see [Cloud storage](/sdks/python/loaders/delta-lake#cloud-storage)), like `'./data/blocks'`.                                                                                             |
| `storage_options`      | dict            | `{}`     | Options and credentials for cloud storage, like `{'AWS_REGION': 'us-east-1'}`.                                                                                                                    |
| `partition_by`         | list of strings | none     | Columns to partition the table by, like `['network']`: rows with the same values go in the same directory. Every load must have these columns, and an existing table must be partitioned by them. |
| `optimize_after_write` | boolean         | `True`   | Whether to compact the table's small files after each write. Turn it off when you stream, which writes often: compacting after every batch slows loading down.                                    |
| `vacuum_after_write`   | boolean         | `False`  | Meant to delete the files that no version of the last 7 days uses, after each write. With the current versions of `deltalake`, it only lists them, and deletes nothing.                           |
| `merge_schema`         | boolean         | `True`   | Whether a load can add columns the table doesn't have yet. Leave it on: turning it off makes every load fail (see [Troubleshooting](/sdks/python/loaders/delta-lake#invalid-schema-write-mode-provided-strict)).                 |
| `schema_evolution`     | boolean         | `True`   | Whether undoing a reorg, which rewrites the table, may replace the table's schema with the one of the rows it keeps.                                                                              |

The `state` and `resilience` keys, which every loader takes, are on [Loaders](/sdks/python/loaders).

Three more keys, `file_size_hint`, `max_rows_per_file`, and `max_rows_per_group`, are meant to size the files the
loader writes, but the current versions of `deltalake` don't accept them: any of them makes every load fail. Leave them
out.

### Load options

`load()` passes `writer_properties` to Delta Lake, to set how it writes Parquet files, like their compression:

```python
from deltalake import WriterProperties

client.sql(query).load(
    connection='local_delta', destination='blocks', read_all=True,
    writer_properties=WriterProperties(compression='ZSTD'),
)
```

## Load modes

Pass `mode` to `load()` ([Loaders](/sdks/python/loaders#load-modes) explains the modes every loader shares):

* **`append`** (the default) adds the rows as a new version of the table.
* **`overwrite`** is meant to replace the table's rows, but each batch Amp sends replaces the one before it, so only the
  last one stays, even with `read_all=True` and a large `batch_size`. The result still counts every row: loading the
  Quickstart's 17,153 transactions this way prints `✅ Loaded 17153 rows` and keeps 769 of them.

`merge` and `upsert` aren't supported: they fail with `Invalid save mode provided: merge`.

To replace the table's rows, delete them, then load in `append` mode. This script does it with the `local_delta`
connection from [Load data](/sdks/python/loaders/delta-lake#load-data):

```python [replace_delta_lake.py]
from deltalake import DeltaTable

from amp import Client

client = Client('grpc://localhost:1602')

if DeltaTable.is_deltatable('./data/blocks'):
    DeltaTable('./data/blocks').delete()

result = client.sql(
    'SELECT block_num, timestamp, hash, gas_used FROM "acme/eth_mainnet".blocks'
).load(connection='local_delta', destination='blocks', read_all=True)

print(result)
```

```bash
python replace_delta_lake.py
```

`delete()` removes every row as a new version, so the previous versions stay readable, and readers see an empty table
until the load writes the next one. For a table in cloud storage, pass the connection's `storage_options` to
`is_deltatable()` and `DeltaTable()` too.

## 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 (see [Streaming](/sdks/python/streaming)). Each batch is a new version of the table. Streamed rows
get one more column, `_amp_batch_id`, which identifies the block range each row came from. When the table doesn't have
it yet, the loader adds it. This stream loads into a new table, `./data/blocks_live`:

```python
client.configure_connection('live_delta', 'deltalake', {'table_path': './data/blocks_live', 'optimize_after_write': False})

results = client.sql(
    'SELECT block_num, timestamp, hash, gas_used FROM "acme/eth_mainnet".blocks'
).load(connection='live_delta', destination='blocks_live', stream=True, with_reorg_detection=False)
for result in results:
    print(result)
```

`with_reorg_detection=False` works around a problem with reorg detection, explained in
[Known issues](/sdks/python/streaming#known-issues).

When Amp reports a [reorg](/get-started/concepts/reorgs), the loader looks up the batches it loaded from the reorged
blocks onward, reads the whole table into memory, removes those batches' rows, and writes what's left as a new version.
That takes longer, and more memory, the larger the table. The version before the reorg stays readable.

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.

## Parallel loads (Unstable)

:::warning
In a [parallel load](/sdks/python/parallel-streaming) into Delta Lake, set `num_workers=1`. With several workers:

* **Rows are repeated, even when every partition succeeds.** Each worker compacts the table after it writes, and
  compactions that overlap keep the same rows more than once. Loading the Quickstart's 17,153 transactions with four
  workers can leave more than 150,000 rows in the table.
* **Partitions can fail on a new table.** The workers race to create it, and the losers fail with
  `Failed to commit transaction: Protocol changed since last commit` or `Metadata changed since last commit`. Their
  rows are missing, and the loop gets a failed result for each, not an exception.

After a load with several workers, [delete the table's rows](/sdks/python/loaders/delta-lake#load-modes), then load the range again with one worker.
:::

## Troubleshooting

### `Invalid save mode provided: merge, only these are supported: ['append', 'overwrite', 'error', 'ignore']`

You passed `mode='merge'` or `mode='upsert'`, which the Delta Lake loader doesn't support. Use `append`, or replace
the table's rows as [Load modes](/sdks/python/loaders/delta-lake#load-modes) shows.

### `Invalid schema write mode provided: strict`

`merge_schema` is `False`, which the current versions of `deltalake` don't support. Remove it from the configuration.

### `write_deltalake() got an unexpected keyword argument 'max_rows_per_file'`

The configuration has `file_size_hint`, `max_rows_per_file`, or `max_rows_per_group`, which the current versions of
`deltalake` don't accept. Remove them. The same error, with `engine` or `large_dtypes`, comes from passing those to
`load()`.

### `Invalid DeltaLakeLoader configuration: DeltaStorageConfig.__init__() got an unexpected keyword argument 'batch_size'`

The configuration has a key the loader doesn't know, like `batch_size`. Pass `batch_size` to `load()` instead, and check
the other keys against [Configuration](/sdks/python/loaders/delta-lake#configuration).

### `Partition columns {'network'} not found in table`

`partition_by` names a column your query doesn't return. Add it to the query, or remove it from `partition_by`.

### `Specified table partitioning does not match table partitioning`

The table already exists, partitioned by other columns than `partition_by`'s. Set `partition_by` to the table's
partition columns, or load into a new `table_path`.

### `Missing S3 credentials: ['AWS_ACCESS_KEY_ID', 'AWS_SECRET_ACCESS_KEY']`

A warning: the loader didn't find the credentials in `storage_options` or the environment. If the load then fails with
`no credentials found in chain`, set them (see [Cloud storage](/sdks/python/loaders/delta-lake#cloud-storage)).

### `Failed to commit transaction: Protocol changed since last commit`

Several workers of a parallel load raced to create the table. `Metadata changed since last commit` has the same cause.
Delete the table's rows, then load with one worker, as [Parallel loads](/sdks/python/loaders/delta-lake#parallel-loads) says.