Iceberg source

This is an experimental feature and is not recommended for use in a production environment. For updates on the progress of the feature or if you want to leave feedback, see the associated GitHub issue.

The iceberg source provides change data capture (CDC) on Apache Iceberg tables. It polls for new Iceberg snapshots and detects INSERT, UPDATE, and DELETE operations by computing changes between snapshots using the IncrementalChangelogScan API.

The source supports the following:

• Iceberg table format v1, v2, and v3.
• Copy-on-Write (CoW) tables only. Merge-on-Read (MoR) tables are not supported.
• Parquet, Avro, and ORC file formats.
• Any Iceberg catalog implementation, including AWS Glue Data Catalog, Hive Metastore, and REST catalogs.

The source includes two ingestion modes:

1. Export: An initial snapshot load that reads the current state of the table.
2. Stream: Polls for new snapshots and processes incremental changelog events.

Prerequisites

To use the Iceberg source, you must enable the experimental iceberg plugin in data-prepper-config.yaml:

experimental:
  enabled_plugins:
    source:
      - iceberg

Usage

The following example pipeline specifies an iceberg source. It reads incremental changes from an Iceberg table using a REST catalog with Amazon S3 storage. The catalog is defined at the top level so that all tables share the same catalog configuration.

In the sink configuration, action is set to ${getMetadata("bulk_action")} so that INSERT and UPDATE operations upsert the document while DELETE operations remove it. The document_id is set to ${getMetadata("document_id")}, which concatenates the identifier_columns values using a | separator so that each Iceberg row maps to a unique OpenSearch document:

version: "2"
iceberg-cdc-pipeline:
  source:
    iceberg:
      catalog:
        type: rest
        uri: "http://iceberg-rest-catalog:8181"
        io-impl: "org.apache.iceberg.aws.s3.S3FileIO"
        client.region: "us-east-1"
      tables:
        - table_name: "my_database.my_table"
          identifier_columns: ["id"]
      polling_interval: "PT30S"
      acknowledgments: true
  sink:
    - opensearch:
        hosts: ["https://localhost:9200"]
        index: "my-index"
        action: "${getMetadata(\"bulk_action\")}"
        document_id: "${getMetadata(\"document_id\")}"

To use a different catalog for a specific table, specify catalog at the table level to override the top-level definition:

iceberg:
  catalog:
    type: rest
    uri: "http://iceberg-rest-catalog:8181"
    io-impl: "org.apache.iceberg.aws.s3.S3FileIO"
  tables:
    - table_name: "db.table_a"
      identifier_columns: ["id"]
    - table_name: "db.table_b"
      identifier_columns: ["id"]
      catalog:
        type: glue
        warehouse: "s3://other-bucket/warehouse"
        io-impl: "org.apache.iceberg.aws.s3.S3FileIO"

Configuration options

The iceberg source supports the following configuration options.

tables

Use the following options for each table entry.

catalog

The catalog option accepts any property supported by the Apache Iceberg catalog implementation. For the full list of available properties, see the documentation for each catalog type, such as the REST catalog, AWS Glue, or Hive Metastore. The following table lists the common properties.

For AWS-specific catalog and S3 properties, see the Apache Iceberg AWS integration documentation.

shuffle

When processing snapshots that contain DELETE operations (UPDATE or DELETE in Copy-on-Write tables), the source uses a distributed shuffle to ensure correct results across multiple nodes. The shuffle routes all records containing the same identifier_columns values to the same node so that matching INSERT and DELETE pairs can be resolved correctly.

The shuffle activates automatically when a snapshot contains DELETE operations. For INSERT-only snapshots, the shuffle is not used.

The following table lists the shuffle configuration options.

When running multiple Data Prepper nodes, each node must be able to reach the other nodes on the configured port.

Exposed metadata attributes

The following metadata attributes are added to each event processed by the iceberg source. These attributes can be accessed using the expression syntax getMetadata function:

• bulk_action: The recommended bulk action for the OpenSearch sink. Set to index for INSERT and UPDATE operations, or delete for DELETE operations.
• document_id: The document identifier, composed by joining the identifier_columns values with a | separator. Only set when identifier_columns is configured.
• iceberg_operation: The CDC operation type. Set to INSERT or DELETE. When a row is updated in Iceberg, the source receives a DELETE and INSERT pair with the same identifier_columns values. The source recognizes this as an update and emits only the INSERT event, because upserting by document_id produces the correct result without a separate delete.
• iceberg_table_name: The name of the source Iceberg table.
• iceberg_snapshot_id: The snapshot ID from which the event originated.

Permissions

The required permissions depend on the catalog type and storage backend. The following is an example of the minimum IAM permissions when using AWS Glue Data Catalog and Amazon S3:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "allowReadingFromS3",
      "Effect": "Allow",
      "Action": [
        "s3:GetObject",
        "s3:ListBucket",
        "s3:GetBucketLocation"
      ],
      "Resource": [
        "arn:aws:s3:::my-iceberg-bucket",
        "arn:aws:s3:::my-iceberg-bucket/*"
      ]
    },
    {
      "Sid": "allowGlueCatalogAccess",
      "Effect": "Allow",
      "Action": [
        "glue:GetTable",
        "glue:GetTables",
        "glue:GetDatabase",
        "glue:GetDatabases"
      ],
      "Resource": [
        "arn:aws:glue:region:account-id:catalog",
        "arn:aws:glue:region:account-id:database/*",
        "arn:aws:glue:region:account-id:table/*/*"
      ]
    }
  ]
}