> **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. # Iceberg Loader The Iceberg loader writes query results into an [Apache Iceberg](https://iceberg.apache.org) table. Iceberg is an open table format for data lakes: Parquet files, plus metadata that tracks every snapshot of the table. A catalog keeps track of the tables, like a REST catalog, AWS Glue, Hive, or a SQL database. Choose it to make chain data available to the engines that read Iceberg, like Spark, Trino, Snowflake, or DuckDB. The loader creates the table for you, and every write is a new snapshot. ## Prerequisites * **The Python SDK**, [installed](/sdks/python#installation), and PyIceberg, the Iceberg library it uses. This page's example uses a catalog in a SQLite database, which needs PyIceberg's `sql-sqlite` extra: :::code-group ```bash [pip] pip install "pyiceberg[sql-sqlite]" ``` ```bash [uv] uv add "pyiceberg[sql-sqlite]" ``` ::: Other catalogs need other extras, like `glue` or `hive`. A REST catalog needs none. Partitioning by a transform, like the day of a timestamp, needs `pyiceberg-core` too. [PyIceberg's documentation](https://py.iceberg.apache.org) lists them. * **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 an Iceberg catalog in `./warehouse`, `local_iceberg`, then loads the Quickstart's 100 blocks into an `eth.blocks` table: ```python [load_iceberg.py] from pathlib import Path from amp import Client warehouse = Path('warehouse').resolve() warehouse.mkdir(exist_ok=True) client = Client('grpc://localhost:1602') client.configure_connection( 'local_iceberg', 'iceberg', { 'catalog_config': { 'name': 'local', 'type': 'sql', 'uri': f'sqlite:///{warehouse}/catalog.db', 'warehouse': f'file://{warehouse}', }, 'namespace': 'eth', }, ) result = client.sql( 'SELECT block_num, timestamp, hash, gas_used FROM "acme/eth_mainnet".blocks' ).load(connection='local_iceberg', destination='blocks', read_all=True) print(result) ``` ```bash python load_iceberg.py ``` * `catalog_config` tells PyIceberg how to reach the catalog. Here, it's a SQLite database, `catalog.db`, and the table's files go in the same directory. * `namespace` groups tables, like a database schema. The loader creates it if it doesn't exist. * `destination` is the table's name in the namespace: the table is `eth.blocks`. * `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, through the same catalog: ```python [read_iceberg.py] from pathlib import Path from pyiceberg.catalog import load_catalog warehouse = Path('warehouse').resolve() catalog = load_catalog('local', type='sql', uri=f'sqlite:///{warehouse}/catalog.db', warehouse=f'file://{warehouse}') table = catalog.load_table('eth.blocks') print(table.schema()) print(table.scan(selected_fields=('block_num', 'gas_used')).to_arrow().sort_by([('block_num', 'descending')]).slice(0, 3)) ``` :::terminal ```bash python read_iceberg.py ``` ``` table { 1: block_num: required long 2: timestamp: optional timestamptz 3: hash: required fixed[32] 4: gas_used: required long } pyarrow.Table block_num: uint64 not null gas_used: uint64 not null ---- block_num: [[24000099,24000098,24000097]] gas_used: [[10751338,36233422,52877862]] ``` ::: ## The table it creates When the table doesn't exist, the loader creates it from your query's columns. Iceberg has no unsigned integers, so unsigned 64-bit columns, like `block_num`, are `long` in the table's schema. Timestamps are stored to the microsecond. The loader doesn't partition the tables it creates, and doesn't give them identifier fields, which [upserts](/sdks/python/loaders/iceberg#load-modes) need. To have either, create the table yourself with PyIceberg, then load into it: the loader uses an existing table as it is. For example, with the `catalog` of `read_iceberg.py`, a table partitioned by day, with `block_num` as its identifier field (the day transform needs `pyiceberg-core`): ```python from pyiceberg.partitioning import PartitionField, PartitionSpec from pyiceberg.schema import Schema from pyiceberg.transforms import DayTransform from pyiceberg.types import LongType, NestedField, TimestamptzType schema = Schema( NestedField(1, 'block_num', LongType(), required=True), NestedField(2, 'timestamp', TimestamptzType(), required=False), NestedField(3, 'gas_used', LongType(), required=True), identifier_field_ids=[1], ) partition_spec = PartitionSpec(PartitionField(source_id=2, field_id=1000, transform=DayTransform(), name='day')) catalog.create_table('eth.blocks_by_day', schema=schema, partition_spec=partition_spec) ``` Then load the same columns into it, like `SELECT block_num, timestamp, gas_used FROM "acme/eth_mainnet".blocks`, with `destination='blocks_by_day'`. ## Configuration The third argument of `configure_connection()`: | Key | Type | Default | Description | | -------------------- | ------- | -------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `catalog_config` | dict | required | How to reach the catalog: its `name`, plus the [catalog properties](https://py.iceberg.apache.org/configuration/) PyIceberg takes, like `type`, `uri`, `warehouse`, and credentials. | | `namespace` | string | required | The namespace of the tables to load into, like `'eth'`. | | `create_namespace` | boolean | `True` | Whether to create the namespace if it doesn't exist. When `False`, a missing namespace fails the load. | | `create_table` | boolean | `True` | Whether to create the table if it doesn't exist. When `False`, a missing table fails the load. | | `schema_evolution` | boolean | `True` | Whether a load can add columns the table doesn't have yet, as optional columns. When `False`, a load with a new column fails. | | `partition_spec` | object | none | A PyIceberg `PartitionSpec` for the tables the loader creates. `configure_connection()` can't save it (see [Troubleshooting](/sdks/python/loaders/iceberg#object-of-type-partitionspec-is-not-json-serializable)): create the table yourself instead. | | `batch_size` | integer | `10000` | Not used. Pass `batch_size` to `load()` instead (see [Loaders](/sdks/python/loaders)). | | `default_table_name` | string | none | Not used. | The `state` and `resilience` keys, which every loader takes, are on [Loaders](/sdks/python/loaders). For a REST catalog, for example, with its token in the `ICEBERG_TOKEN` environment variable: ```python import os client.configure_connection( 'rest_iceberg', 'iceberg', { 'catalog_config': { 'name': 'rest', 'type': 'rest', 'uri': 'https://catalog.example.com', 'token': os.environ['ICEBERG_TOKEN'], }, 'namespace': 'eth', }, ) ``` `configure_connection()` saves the connection, token included, so later scripts can use it without configuring it again. [Loaders](/sdks/python/loaders) explains where it saves it. ## 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 snapshot. * **`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. * **`upsert`** and **`merge`**, which do the same, update the rows whose identifier fields match a row you load, and add the others. The table needs identifier fields, which the tables the loader creates don't have (see [The table it creates](/sdks/python/loaders/iceberg#the-table-it-creates)). Without them, the loader logs `UPSERT/MERGE operation failed: Join columns could not be found` and appends the rows instead, duplicating the ones the table already has. To replace the table's rows, delete them, then load in `append` mode. This script does it with the `local_iceberg` connection from [Load data](/sdks/python/loaders/iceberg#load-data): ```python [replace_iceberg.py] from pathlib import Path from pyiceberg.catalog import load_catalog from amp import Client warehouse = Path('warehouse').resolve() catalog = load_catalog('local', type='sql', uri=f'sqlite:///{warehouse}/catalog.db', warehouse=f'file://{warehouse}') if catalog.table_exists('eth.blocks'): catalog.load_table('eth.blocks').delete() client = Client('grpc://localhost:1602') result = client.sql( 'SELECT block_num, timestamp, hash, gas_used FROM "acme/eth_mainnet".blocks' ).load(connection='local_iceberg', destination='blocks', read_all=True) print(result) ``` ```bash python replace_iceberg.py ``` `delete()` removes every row as a new snapshot, so the previous snapshots stay in the table's history, and readers see an empty table until the load writes the next one. ## 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 snapshot. 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 as an optional column, unless `schema_evolution` is `False`. ```python results = client.sql( 'SELECT block_num, timestamp, hash, gas_used FROM "acme/eth_mainnet".blocks' ).load(connection='local_iceberg', 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 overwrites the table with what's left, as a new snapshot. That takes longer, and more memory, the larger the table. The snapshot before the reorg stays in the table's history. 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 Iceberg, set `num_workers=1`. Parallel loads add the `_amp_batch_id` column to the table, and with several workers, on a table without it, like a new one, they all try to at once. Most partitions then fail with `Failed to evolve table schema: Table has been updated by another process`. Their rows are missing, and the loop gets a failed result for each, not an exception. Loading the Quickstart's 17,153 transactions with four workers into a new table loads one partition, about a quarter of the rows. The partitions that succeeded are already in the table, so [delete its rows](/sdks/python/loaders/iceberg#load-modes) before you load the range again with one worker. ::: ## Troubleshooting ### `Object of type PartitionSpec is not JSON serializable` `configure_connection()` saves connections as JSON, which can't hold a `PartitionSpec`. Remove `partition_spec`, and create the partitioned table yourself (see [The table it creates](/sdks/python/loaders/iceberg#the-table-it-creates)). The failed save also leaves the file of saved connections incomplete: configure your other connections again. ### `UPSERT/MERGE operation failed: Join columns could not be found, please set identifier-field-ids or pass in explicitly.. Falling back to APPEND mode.` The table has no identifier fields, so the loader appended the rows. Create the table with identifier fields (see [The table it creates](/sdks/python/loaders/iceberg#the-table-it-creates)), or use `append`. ### `Table 'eth.blocks' not found and create_table=False` `create_table` is `False` in the configuration, and the table doesn't exist. Create it, or remove `create_table`. ### `Failed to verify namespace 'eth': Namespace 'eth' not found` `create_namespace` is `False`, and the namespace doesn't exist. Create it, or remove `create_namespace`. ### `Arrow schema contains columns not present in Iceberg table` `schema_evolution` is `False`, and your query returns a column the table doesn't have, like `_amp_batch_id` when you stream. Add the column to the table, remove it from your query, or remove `schema_evolution`. ### `pyiceberg_core needs to be installed` The table is partitioned by a transform, like the day of a timestamp, which needs `pyiceberg-core`. Install it, like `pip install "pyiceberg[sql-sqlite,pyiceberg-core]"`. ### `Failed to evolve table schema: Table has been updated by another process` Several workers of a parallel load tried to add the `_amp_batch_id` column at once. `Failed to evolve table schema: Requirement failed: current schema id has changed` has the same cause. Delete the table's rows, then load with one worker, as [Parallel loads](/sdks/python/loaders/iceberg#parallel-loads) says.