---
title: InfluxDB to Iceberg plugin
description: Export time series data from InfluxDB to Apache Iceberg table format for data lake integration.
url: https://docs.influxdata.com/influxdb3/core/plugins/library/official/influxdb-to-iceberg/
estimated_tokens: 4609
product: InfluxDB 3 Core
version: core
publisher: InfluxData
canonical: https://docs.influxdata.com/influxdb3/core/plugins/library/official/influxdb-to-iceberg/
date: '2026-07-11T10:53:18+00:00'
lastmod: '2026-07-11T10:53:18+00:00'
---

The InfluxDB to Iceberg Plugin enables data transfer from InfluxDB 3 Core to Apache Iceberg tables. Transfer time series data to Iceberg for long-term storage, analytics, or integration with data lake architectures. The plugin supports both scheduled batch transfers of historical data and on-demand transfers via HTTP API.

## Configuration

Plugin parameters may be specified as key-value pairs in the `--trigger-arguments` flag (CLI) or in the `trigger_arguments` field (API) when creating a trigger. Some plugins support TOML configuration files, which can be specified using the plugin’s `config_file_path` parameter.

If a plugin supports multiple trigger specifications, some parameters may depend on the trigger specification that you use.

### Plugin metadata

This plugin includes a JSON metadata schema in its docstring that defines supported trigger types and configuration parameters. This metadata enables the [InfluxDB 3 Explorer](https://docs.influxdata.com/influxdb3/explorer/) UI to display and configure the plugin.

### Scheduler trigger parameters

#### Required parameters

|    Parameter    | Type |Default |                                   Description                                    |
|-----------------|------|--------|----------------------------------------------------------------------------------|
|  `measurement`  |string|required|                  Source measurement containing data to transfer                  |
|    `window`     |string|required|Time window for data transfer. Format: `<number><unit>` (for example, “1h”, “30d”)|
|`catalog_configs`|string|required|       Base64-encoded JSON string containing Iceberg catalog configuration        |

#### Optional parameters

|     Parameter      | Type |    Default     |                                   Description                                    |
|--------------------|------|----------------|----------------------------------------------------------------------------------|
| `included_fields`  |string|all fields/tags |Dot-separated list of fields and tags to include (for example, “usage\_user.host”)|
| `excluded_fields`  |string|      none      |                 Dot-separated list of fields and tags to exclude                 |
|    `namespace`     |string|   “default”    |                      Iceberg namespace for the target table                      |
|    `table_name`    |string|measurement name|                                Iceberg table name                                |
|`auto_update_schema`|string|     false      |Automatically update Iceberg table schema when data doesn’t match existing schema |

### TOML configuration

|    Parameter     | Type |Default|                                  Description                                   |
|------------------|------|-------|--------------------------------------------------------------------------------|
|`config_file_path`|string| none  |TOML config file path relative to `PLUGIN_DIR` (required for TOML configuration)|

*To use a TOML configuration file, set the `PLUGIN_DIR` environment variable and specify the `config_file_path` in the trigger arguments.* This is in addition to the `--plugin-dir` flag when starting InfluxDB 3 Core.

#### Example TOML configuration

[influxdb\_to\_iceberg\_config\_scheduler.toml](https://github.com/influxdata/influxdb3_plugins/blob/master/influxdata/influxdb_to_iceberg/influxdb_to_iceberg_config_scheduler.toml)

For more information on using TOML configuration files, see the Using TOML Configuration Files section in the [influxdb3\_plugins/README.md](https://github.com/influxdata/influxdb3_plugins/blob/master/README.md).

### HTTP trigger parameters

#### Request body structure

|     Parameter      | Type  |Required|                                                         Description                                                         |
|--------------------|-------|--------|-----------------------------------------------------------------------------------------------------------------------------|
|   `measurement`    |string |  Yes   |                                       Source measurement containing data to transfer                                        |
| `catalog_configs`  |object |  Yes   |Iceberg catalog configuration dictionary. See [PyIceberg catalog documentation](https://py.iceberg.apache.org/configuration/)|
| `included_fields`  | array |   No   |                                    List of field and tag names to include in replication                                    |
| `excluded_fields`  | array |   No   |                                   List of field and tag names to exclude from replication                                   |
|    `namespace`     |string |   No   |                                        Target Iceberg namespace (default: “default”)                                        |
|    `table_name`    |string |   No   |                                    Target Iceberg table name (default: measurement name)                                    |
|    `batch_size`    |string |   No   |                        Batch size duration for processing (default: “1d”). Format: `<number><unit>`                         |
|  `backfill_start`  |string |   No   |                                     ISO 8601 datetime with timezone for backfill start                                      |
|   `backfill_end`   |string |   No   |                                      ISO 8601 datetime with timezone for backfill end                                       |
|`auto_update_schema`|boolean|   No   |             Automatically update Iceberg table schema when data doesn’t match existing schema (default: false)              |

## Schema management

* Automatically creates Iceberg table schema from the first batch of data
* Maps pandas data types to Iceberg types:
  * `int64` → `IntegerType`
  * `float64` → `FloatType`
  * `datetime64[us]` → `TimestampType`
  * `object` → `StringType`

* Fields with no null values are marked as `required`
* The `time` column is converted to `datetime64[us]` for Iceberg compatibility
* Tables are created in format: `<namespace>.<table_name>`

### Automatic schema updates

When `auto_update_schema=true`:

* **New fields**: Automatically added to Iceberg table schema as optional (nullable) columns
* **Missing fields**: Added to DataFrame with null values based on existing schema types
* **Schema evolution**: Ensures data compatibility between InfluxDB and Iceberg without manual intervention
* **Backward compatibility**: Existing data remains valid as new columns are always optional

## Software Requirements

* **InfluxDB 3 Core**: with the Processing Engine enabled
* **Python packages**:
  * `pandas` (for data manipulation)
  * `pyarrow` (for Parquet support)
  * `pyiceberg[catalog-options]` (for Iceberg integration)

### Installation steps

1. Start InfluxDB 3 Core with the Processing Engine enabled (`--plugin-dir /path/to/plugins`):

   ```
   influxdb3 serve \
     --node-id node0 \
     --object-store file \
     --data-dir ~/.influxdb3 \
     --plugin-dir ~/.plugins
   ```

2. Install required Python packages:

   ```
   influxdb3 install package pandas
   influxdb3 install package pyarrow
   influxdb3 install package "pyiceberg[s3fs,hive,sql-sqlite]"
   ```

**Note:** Include the appropriate PyIceberg extras based on your catalog type:

* `[s3fs]` for S3 storage
* `[hive]` for Hive metastore
* `[sql-sqlite]` for SQL catalog with SQLite
* See [PyIceberg documentation](https://py.iceberg.apache.org/#installation) for all options

## Schema requirement

The plugin assumes that the table schema is already defined in the database, as it relies on this schema to retrieve field and tag names required for processing.

## Trigger setup

### Scheduled data transfer

Periodically transfer data from InfluxDB 3 Core to Iceberg:

```bash
influxdb3 create trigger \
  --database mydb \
  --path "gh:influxdata/influxdb_to_iceberg/influxdb_to_iceberg.py" \
  --trigger-spec "every:1h" \
  --trigger-arguments 'measurement=cpu,window=1h,catalog_configs="eyJ1cmkiOiAiaHR0cDovL25lc3NpZTo5MDAwIn0=",namespace=monitoring,table_name=cpu_metrics' \
  hourly_iceberg_transfer
```

### HTTP API endpoint

Create an on-demand transfer endpoint:

```bash
influxdb3 create trigger \
  --database mydb \
  --path "gh:influxdata/influxdb_to_iceberg/influxdb_to_iceberg.py" \
  --trigger-spec "request:replicate" \
  iceberg_http_transfer
```

Enable the trigger:

```bash
influxdb3 enable trigger --database mydb iceberg_http_transfer
```

The endpoint is registered at `/api/v3/engine/replicate`.

## Example usage

### Example 1: Basic scheduled transfer

Transfer CPU metrics to Iceberg every hour:

```bash
# Create trigger with base64-encoded catalog config
# Original JSON: {"uri": "http://nessie:9000"}
# Base64: eyJ1cmkiOiAiaHR0cDovL25lc3NpZTo5MDAwIn0=
influxdb3 create trigger \
  --database metrics \
  --path "gh:influxdata/influxdb_to_iceberg/influxdb_to_iceberg.py" \
  --trigger-spec "every:1h" \
  --trigger-arguments 'measurement=cpu,window=24h,catalog_configs="eyJ1cmkiOiAiaHR0cDovL25lc3NpZTo5MDAwIn0="' \
  cpu_to_iceberg

# Write test data
influxdb3 write \
  --database metrics \
  "cpu,host=server1 usage_user=45.2,usage_system=12.1"

# After trigger runs, data is available in Iceberg table "default.cpu"
```

**Expected output**

* Creates Iceberg table `default.cpu` with schema matching the measurement
* Transfers all CPU data from the last 24 hours
* Appends new data on each hourly run

### Example 2: HTTP backfill with field filtering

Backfill specific fields from historical data:

```bash
# Create and enable HTTP trigger
influxdb3 create trigger \
  --database metrics \
  --path "gh:influxdata/influxdb_to_iceberg/influxdb_to_iceberg.py" \
  --trigger-spec "request:replicate" \
  iceberg_backfill

influxdb3 enable trigger --database metrics iceberg_backfill

# Request backfill via HTTP
curl -X POST http://localhost:8181/api/v3/engine/replicate \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -d '{
    "measurement": "temperature",
    "catalog_configs": {
      "type": "sql",
      "uri": "sqlite:///path/to/catalog.db"
    },
    "included_fields": ["temp_celsius", "humidity", "sensor_id"],
    "namespace": "weather",
    "table_name": "temperature_history",
    "batch_size": "12h",
    "backfill_start": "2024-01-01T00:00:00+00:00",
    "backfill_end": "2024-01-07T00:00:00+00:00"
  }'
```

**Expected output**

* Creates Iceberg table `weather.temperature_history`
* Transfers only `temp_celsius` and `humidity` fields
* Processes data in 12-hour batches for the specified week
* Returns status of the backfill operation

### Example 3: S3-backed Iceberg catalog

Transfer data to Iceberg tables stored in S3:

```bash
# Create catalog config JSON
cat > catalog_config.json << EOF
{
  "type": "sql",
  "uri": "sqlite:///iceberg/catalog.db",
  "warehouse": "s3://my-bucket/iceberg-warehouse/",
  "s3.endpoint": "http://minio:9000",
  "s3.access-key-id": "minioadmin",
  "s3.secret-access-key": "minioadmin",
  "s3.path-style-access": true
}
EOF

# Encode to base64
CATALOG_CONFIG=$(base64 < catalog_config.json)

# Create trigger
influxdb3 create trigger \
  --database metrics \
  --path "gh:influxdata/influxdb_to_iceberg/influxdb_to_iceberg.py" \
  --trigger-spec "every:30m" \
  --trigger-arguments "measurement=sensor_data,window=1h,catalog_configs=\"$CATALOG_CONFIG\",namespace=iot,table_name=sensors" \
  s3_iceberg_transfer
```

## Code overview

### Files

* `influxdb_to_iceberg.py`: The main plugin code containing handlers for scheduled and HTTP triggers
* `influxdb_to_iceberg_config_scheduler.toml`: Example TOML configuration file for scheduled triggers

### Logging

Logs are stored in the trigger’s database in the `system.processing_engine_logs` table. To view logs:

```bash
influxdb3 query --database YOUR_DATABASE "SELECT * FROM system.processing_engine_logs WHERE trigger_name = 'your_trigger_name'"
```

Log columns:

* **event\_time**: Timestamp of the log event
* **trigger\_name**: Name of the trigger that generated the log
* **log\_level**: Severity level (INFO, WARN, ERROR)
* **log\_text**: Message describing the action or error

### Main functions

#### `process_scheduled_call(influxdb3_local, call_time, args)`

Handles scheduled data transfers. Queries data within the specified window and appends to Iceberg tables.

Key operations:

1. Parses configuration and decodes catalog settings
2. Queries source measurement with optional field filtering
3. Creates Iceberg table if needed
4. Appends data to Iceberg table

#### `process_http_request(influxdb3_local, request_body, args)`

Handles on-demand data transfers via HTTP. Supports backfill operations with configurable batch sizes.

Key operations:

1. Validates request body parameters
2. Determines backfill time range
3. Processes data in batches
4. Returns transfer status

## Troubleshooting

### Common issues

#### Issue: “Failed to decode catalog\_configs” error

**Solution**: Ensure the catalog configuration is properly base64-encoded:

```bash
# Create JSON file
echo '{"uri": "http://nessie:9000"}' > config.json
# Encode to base64
base64 config.json
```

#### Issue: “Failed to create Iceberg table” error

**Solution**:

1. Verify catalog configuration is correct
2. Check warehouse path permissions
3. Ensure required PyIceberg extras are installed:`bash influxdb3 install package "pyiceberg[s3fs]"`

#### Issue: No data in Iceberg table after transfer

**Solution**:

1. Check if source measurement contains data:`bash influxdb3 query --database mydb "SELECT COUNT(*) FROM measurement"`
2. Verify time window covers data:`bash influxdb3 query --database mydb "SELECT MIN(time), MAX(time) FROM measurement"`
3. Check logs for errors:`bash influxdb3 query --database YOUR_DATABASE "SELECT * FROM system.processing_engine_logs WHERE log_level = 'ERROR'"`

#### Issue: “Incompatible change: cannot add required column” error

**Solution**: This occurs when trying to add a required (non-nullable) column to an existing table. With `auto_update_schema=true`, new columns are automatically added as optional. If you encounter this error:

1. Ensure `auto_update_schema=true` in your configuration
2. Check that you’re using the latest version of the plugin

### Debugging tips

1. **Test catalog connectivity**:

```python
from pyiceberg.catalog import load_catalog
catalog = load_catalog("my_catalog", **catalog_configs)
print(catalog.list_namespaces())
```

1. **Verify field names**:

   ```
   influxdb3 query --database mydb "SHOW FIELD KEYS FROM measurement"
   ```

2. **Use smaller windows** for initial testing:

   ```
   --trigger-arguments 'window=5m,...'
   ```

### Performance considerations

* **File sizing**: Each scheduled run creates new Parquet files. Use appropriate window sizes to balance file count and size
* **Batch processing**: For HTTP transfers, adjust `batch_size` based on available memory
* **Field and tag filtering**: Use `included_fields` to reduce data volume when only specific fields and tags are needed
* **Catalog choice**: SQL catalogs (SQLite) are simpler but REST catalogs scale better

## Report an issue

For plugin issues, see the Plugins repository [issues page](https://github.com/influxdata/influxdb3_plugins/issues).

## Find support for InfluxDB 3 Core

The [InfluxDB Discord server](https://discord.gg/9zaNCW2PRT) is the best place to find support for InfluxDB 3 Core and InfluxDB 3 Enterprise.
For other InfluxDB versions, see the [Support and feedback](#bug-reports-and-feedback) options.

#### Related

* [InfluxDB to Iceberg plugin on GitHub](https://github.com/influxdata/influxdb3_plugins/tree/main/influxdata/influxdb_to_iceberg)

[plugins](/influxdb3/core/tags/plugins/)[processing engine](/influxdb3/core/tags/processing-engine/)[python](/influxdb3/core/tags/python/)[iceberg](/influxdb3/core/tags/iceberg/)[export](/influxdb3/core/tags/export/)[data-lake](/influxdb3/core/tags/data-lake/)
| Parameter | Type | Default | Description |
| --- | --- | --- | --- |
| Parameter | Type | Default | Description |
| measurement | string | required | Source measurement containing data to transfer |
| window | string | required | Time window for data transfer. Format:  <number><unit>  (for example, “1h”, “30d”) |
| catalog_configs | string | required | Base64-encoded JSON string containing Iceberg catalog configuration |

| Parameter | Type | Default | Description |
| --- | --- | --- | --- |
| Parameter | Type | Default | Description |
| included_fields | string | all fields/tags | Dot-separated list of fields and tags to include (for example, “usage_user.host”) |
| excluded_fields | string | none | Dot-separated list of fields and tags to exclude |
| namespace | string | “default” | Iceberg namespace for the target table |
| table_name | string | measurement name | Iceberg table name |
| auto_update_schema | string | false | Automatically update Iceberg table schema when data doesn’t match existing schema |

| Parameter | Type | Default | Description |
| --- | --- | --- | --- |
| Parameter | Type | Default | Description |
| config_file_path | string | none | TOML config file path relative to  PLUGIN_DIR  (required for TOML configuration) |

| Parameter | Type | Required | Description |
| --- | --- | --- | --- |
| Parameter | Type | Required | Description |
| measurement | string | Yes | Source measurement containing data to transfer |
| catalog_configs | object | Yes | Iceberg catalog configuration dictionary. See  PyIceberg catalog documentation |
| included_fields | array | No | List of field and tag names to include in replication |
| excluded_fields | array | No | List of field and tag names to exclude from replication |
| namespace | string | No | Target Iceberg namespace (default: “default”) |
| table_name | string | No | Target Iceberg table name (default: measurement name) |
| batch_size | string | No | Batch size duration for processing (default: “1d”). Format:  <number><unit> |
| backfill_start | string | No | ISO 8601 datetime with timezone for backfill start |
| backfill_end | string | No | ISO 8601 datetime with timezone for backfill end |
| auto_update_schema | boolean | No | Automatically update Iceberg table schema when data doesn’t match existing schema (default: false) |