Filesystem to Iceberg

The complete flow uses the public HTTP API:

S3 files -> Filesystem source table -> SQL pipeline -> Iceberg sink table

Prerequisites

You need:

the public URL and bearer token for the Ingest Engine API;
an S3 prefix containing newline-delimited JSON files;
an AWS Glue database and an S3 warehouse location; and
an AWS identity that can read the source prefix and access the Glue catalog and warehouse.

Set the API values used by the requests:

export INGESTION_ENGINE_API_URL="https://<ingest-engine-api-host>"
export INGESTION_ENGINE_API_TOKEN="<api-token>"

The examples omit static AWS credentials and use the runtime’s configured AWS credential chain. If your environment requires explicit storage settings, add the options described in the Filesystem and Iceberg references.

1. Create the Filesystem Source Table

Create a typed Filesystem table through POST /api/v1/connectors/filesystem/tables:

curl --fail-with-body -X POST \
  "$INGESTION_ENGINE_API_URL/api/v1/connectors/filesystem/tables" \
  -H "Authorization: Bearer $INGESTION_ENGINE_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "orders_source",
    "config": {
      "table_type": {
        "type": "source",
        "source_config": {
          "path": "s3://<source-bucket>/raw/orders/",
          "compression_format": "none",
          "regex_pattern": ".*\\.jsonl$",
          "storage_options": {
            "region": "us-east-1"
          }
        }
      }
    },
    "schema": {
      "format": {
        "json": {}
      },
      "fields": [
        {
          "field_name": "order_id",
          "field_type": {
            "type": {
              "primitive": "String"
            }
          },
          "nullable": false
        },
        {
          "field_name": "order_time",
          "field_type": {
            "type": {
              "primitive": "DateTime"
            }
          },
          "nullable": false
        },
        {
          "field_name": "amount",
          "field_type": {
            "type": {
              "primitive": "F64"
            }
          },
          "nullable": false
        },
        {
          "field_name": "region",
          "field_type": {
            "type": {
              "primitive": "String"
            }
          },
          "nullable": false
        }
      ]
    }
  }'

The table’s name, orders_source, is the name used in pipeline SQL. The source recursively scans the configured path and includes only object keys that match regex_pattern. Supported source compression values are none, gzip, and zstd. The response contains the table’s generated id; keep it for later API lookups.

2. Create the Iceberg Sink Table

Iceberg configuration is supplied directly on the table and does not use a connection profile. Create the sink through POST /api/v1/connectors/iceberg/tables:

curl --fail-with-body -X POST \
  "$INGESTION_ENGINE_API_URL/api/v1/connectors/iceberg/tables" \
  -H "Authorization: Bearer $INGESTION_ENGINE_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "orders_iceberg_sink",
    "config": {
      "catalog": {
        "provider": "direct",
        "format": "iceberg",
        "type": "glue",
        "region": "us-east-1",
        "warehouse": "s3://<warehouse-bucket>/warehouse"
      },
      "namespace": "analytics",
      "tableName": "orders",
      "appendOnly": true,
      "partitioning": {
        "fields": [
          {
            "name": "region",
            "transform": "identity"
          }
        ],
        "shuffle_by_partition": {
          "enabled": false
        }
      },
      "storageOptions": {},
      "fileRotation": {
        "maxFileSizeBytes": 134217728
      },
      "parquet": {
        "compression": "zstd",
        "compressionLevel": 3,
        "maxRowGroupRows": 1000000,
        "dataPageSize": 1048576,
        "dictionaryEnabled": true
      },
      "uploadChunkSizeBytes": 8388608,
      "uploadConcurrency": 8,
      "iceberg": {
        "targetFileSizeBytes": 268435456
      }
    },
    "schema": {
      "format": {
        "parquet": {}
      },
      "fields": [
        {
          "field_name": "order_id",
          "field_type": {
            "type": {
              "primitive": "String"
            }
          },
          "nullable": false
        },
        {
          "field_name": "order_time",
          "field_type": {
            "type": {
              "primitive": "DateTime"
            }
          },
          "nullable": false
        },
        {
          "field_name": "amount",
          "field_type": {
            "type": {
              "primitive": "F64"
            }
          },
          "nullable": false
        },
        {
          "field_name": "region",
          "field_type": {
            "type": {
              "primitive": "String"
            }
          },
          "nullable": false
        }
      ]
    },
    "environment": "production"
  }'

The destination is analytics.orders in the configured Glue catalog. appendOnly: true selects insert-only output, and the partitioning block writes records into Iceberg partitions based on region. Production table creation requires explicit file, upload, partitioning, and Parquet settings; the payload above includes those settings.

3. Validate the Pipeline Query

Validate the SQL before starting the pipeline:

curl --fail-with-body -X POST \
  "$INGESTION_ENGINE_API_URL/api/v1/pipelines/validate_query" \
  -H "Authorization: Bearer $INGESTION_ENGINE_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "query": "INSERT INTO orders_iceberg_sink SELECT order_id, order_time, amount, UPPER(region) AS region FROM orders_source"
  }'

A valid query returns a planned graph and an empty errors array.

4. Create the Pipeline

Create and start the pipeline through POST /api/v1/pipelines:

curl --fail-with-body -X POST \
  "$INGESTION_ENGINE_API_URL/api/v1/pipelines" \
  -H "Authorization: Bearer $INGESTION_ENGINE_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "orders_to_iceberg",
    "query": "INSERT INTO orders_iceberg_sink SELECT order_id, order_time, amount, UPPER(region) AS region FROM orders_source",
    "parallelism": 1,
    "checkpointIntervalMicros": 10000000,
    "environment": "production"
  }'

The response contains the generated pipeline id. A checkpoint interval of 10000000 is 10 seconds. Iceberg commits become visible after successful checkpoints.

5. Monitor Through the API

Substitute the generated ID returned by the create call:

curl --fail-with-body \
  "$INGESTION_ENGINE_API_URL/api/v1/pipelines/<pipeline-id>" \
  -H "Authorization: Bearer $INGESTION_ENGINE_API_TOKEN"

curl --fail-with-body \
  "$INGESTION_ENGINE_API_URL/api/v1/pipelines/<pipeline-id>/jobs" \
  -H "Authorization: Bearer $INGESTION_ENGINE_API_TOKEN"

The first response returns the pipeline definition and planned graph. The jobs response reports the current execution state.

Continue

Continue with the Architecture Overview to see how the controller, workers, connectors, and checkpoints execute these pipelines.

Ingest Engine

Get started

Connectors

Streaming SQL

Operate

Reference

Concepts

Filesystem to Iceberg

Prerequisites

1. Create the Filesystem Source Table

2. Create the Iceberg Sink Table

3. Validate the Pipeline Query

4. Create the Pipeline

5. Monitor Through the API

Continue

​Prerequisites

​1. Create the Filesystem Source Table

​2. Create the Iceberg Sink Table

​3. Validate the Pipeline Query

​4. Create the Pipeline

​5. Monitor Through the API

​Continue

Prerequisites

1. Create the Filesystem Source Table

2. Create the Iceberg Sink Table

3. Validate the Pipeline Query

4. Create the Pipeline

5. Monitor Through the API

Continue