HTTP to Iceberg

This example uses only the public Ingest Engine API. You will create:

an HTTP source table named webhook_events;
an Iceberg sink table named webhook_events_iceberg; and
a SQL pipeline that connects them.

Prerequisites

You need:

the base URL for your Ingest Engine API;
an Ingest Engine API bearer token;
an Iceberg REST catalog; and
an S3 or S3-compatible warehouse accessible to the Ingest Engine.

Set the API values used by the requests:

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

Replace every value enclosed in <...> before sending the requests. Keep API, catalog, source, and storage credentials out of source control.

1. Create the HTTP source table

Create a source that accepts JSON objects at /webhooks:

curl --fail-with-body --silent --show-error \
  -X POST "$INGESTION_ENGINE_API_URL/api/v1/connectors/http_source/tables" \
  -H "Authorization: Bearer $INGESTION_ENGINE_API_TOKEN" \
  -H "Content-Type: application/json" \
  --data '{
    "name": "webhook_events",
    "config": {
      "port": 8080,
      "path": "/webhooks",
      "bind_address": "0.0.0.0",
      "buffer_size": 1000,
      "max_body_size": 5242880,
      "auth": {
        "type": "bearer",
        "token": "<http-source-token>"
      }
    },
    "schema": {
      "format": {
        "json": {}
      },
      "bad_data": {
        "fail": {}
      },
      "fields": [
        {
          "field_name": "event_id",
          "field_type": {"type": {"primitive": "String"}},
          "nullable": false
        },
        {
          "field_name": "event_type",
          "field_type": {"type": {"primitive": "String"}},
          "nullable": false
        },
        {
          "field_name": "source",
          "field_type": {"type": {"primitive": "String"}},
          "nullable": false
        },
        {
          "field_name": "payload",
          "field_type": {"type": {"primitive": "String"}},
          "nullable": true
        },
        {
          "field_name": "event_time",
          "field_type": {"type": {"primitive": "DateTime"}},
          "nullable": false
        }
      ]
    },
    "environment": "production"
  }'

The response is the created connection table. Save its id if you need to retrieve it later with GET /api/v1/connection_tables/{id}. The source bearer token authenticates event producers. It is separate from the Ingest Engine API token used to create and inspect resources.

2. Create the Iceberg sink table

Create an append-only Iceberg table in the webhooks namespace. This configuration uses a REST catalog and partitions records by the hour of event_time.

curl --fail-with-body --silent --show-error \
  -X POST "$INGESTION_ENGINE_API_URL/api/v1/connectors/iceberg/tables" \
  -H "Authorization: Bearer $INGESTION_ENGINE_API_TOKEN" \
  -H "Content-Type: application/json" \
  --data '{
    "name": "webhook_events_iceberg",
    "config": {
      "catalog": {
        "provider": "direct",
        "format": "iceberg",
        "type": "rest",
        "url": "<iceberg-rest-catalog-url>",
        "warehouse": "s3://<bucket>/warehouse",
        "auth": {
          "token": "<iceberg-catalog-token>"
        }
      },
      "namespace": "webhooks",
      "tableName": "events",
      "appendOnly": true,
      "partitioning": {
        "fields": [
          {
            "name": "event_time",
            "transform": "hour"
          }
        ],
        "shuffle_by_partition": {
          "enabled": true
        }
      },
      "storageOptions": {
        "s3.endpoint": "<s3-endpoint>",
        "s3.region": "<s3-region>",
        "s3.access-key-id": "<s3-access-key-id>",
        "s3.secret-access-key": "<s3-secret-access-key>",
        "s3.path-style-access": "true",
        "s3.disable-config-load": "true"
      },
      "fileRotation": {
        "maxFileSizeBytes": 134217728
      },
      "parquet": {
        "compression": "zstd",
        "compressionLevel": 3,
        "maxRowGroupRows": 1000000,
        "dataPageSize": 1048576,
        "dictionaryEnabled": true,
        "dictionaryPageSize": 1048576
      },
      "uploadChunkSizeBytes": 8388608,
      "uploadConcurrency": 8,
      "iceberg": {
        "targetFileSizeBytes": 268435456
      }
    },
    "schema": {
      "format": {
        "parquet": {}
      },
      "fields": [
        {
          "field_name": "event_id",
          "field_type": {"type": {"primitive": "String"}},
          "nullable": false
        },
        {
          "field_name": "event_type",
          "field_type": {"type": {"primitive": "String"}},
          "nullable": false
        },
        {
          "field_name": "source",
          "field_type": {"type": {"primitive": "String"}},
          "nullable": false
        },
        {
          "field_name": "payload",
          "field_type": {"type": {"primitive": "String"}},
          "nullable": true
        },
        {
          "field_name": "event_time",
          "field_type": {"type": {"primitive": "DateTime"}},
          "nullable": false
        }
      ]
    },
    "environment": "production"
  }'

If the catalog allows anonymous access, omit catalog.auth. If the warehouse uses the default cloud credential chain, remove the static access keys and any custom endpoint settings that are unnecessary for your environment. The response contains the sink connection table and its id.

3. Validate the pipeline query

Table names in the SQL query are the name values from the two table requests.

curl --fail-with-body --silent --show-error \
  -X POST "$INGESTION_ENGINE_API_URL/api/v1/pipelines/validate_query" \
  -H "Authorization: Bearer $INGESTION_ENGINE_API_TOKEN" \
  -H "Content-Type: application/json" \
  --data '{
    "query": "INSERT INTO webhook_events_iceberg SELECT event_id, event_type, source, payload, event_time FROM webhook_events"
  }'

A valid query returns an empty errors array and a non-null graph. Resolve any reported errors before creating the pipeline.

4. Create the pipeline

curl --fail-with-body --silent --show-error \
  -X POST "$INGESTION_ENGINE_API_URL/api/v1/pipelines" \
  -H "Authorization: Bearer $INGESTION_ENGINE_API_TOKEN" \
  -H "Content-Type: application/json" \
  --data '{
    "name": "http-to-iceberg-webhooks",
    "query": "INSERT INTO webhook_events_iceberg SELECT event_id, event_type, source, payload, event_time FROM webhook_events",
    "parallelism": 1,
    "checkpointIntervalMicros": 10000000,
    "environment": "production"
  }'

The response includes the pipeline id, its execution graph, and actionText. Save the id for the monitoring requests. The checkpoint interval is expressed in microseconds; 10000000 is 10 seconds. Iceberg commits occur at checkpoints, so data can become visible after the source request has been accepted.

5. Send an event

Set HTTP_SOURCE_URL to the public base URL assigned to this HTTP source in your Ingest Engine environment. This URL is separate from INGESTION_ENGINE_API_URL and should not be an internal service address.

export HTTP_SOURCE_URL="https://<public-http-source-url>"
export HTTP_SOURCE_TOKEN="<http-source-token>"

curl --fail-with-body --silent --show-error \
  -X POST "$HTTP_SOURCE_URL/webhooks" \
  -H "Authorization: Bearer $HTTP_SOURCE_TOKEN" \
  -H "Content-Type: application/json" \
  --data '{
    "event_id": "evt-001",
    "event_type": "user.signup",
    "source": "auth-service",
    "payload": "{\"user_id\":12345}",
    "event_time": "2026-07-15T10:30:00Z"
  }'

A successful response means the source accepted the event. It does not by itself confirm that the event has been committed to Iceberg.

6. Monitor the pipeline

Retrieve the pipeline using the id returned when it was created:

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

Use actionText and actionInProgress to follow the requested pipeline action. Inspect the pipeline’s jobs to see their runtime state and any failure message:

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

Each job in the response has a state field. If a job fails, failureMessage contains the reported cause.

Continue

Continue with Filesystem to Iceberg for an API-driven pipeline that reads files from object storage.

Ingest Engine

Get started

Connectors

Streaming SQL

Operate

Reference

Concepts

Prerequisites

1. Create the HTTP source table

2. Create the Iceberg sink table

3. Validate the pipeline query

4. Create the pipeline

5. Send an event

6. Monitor the pipeline

Continue

​Prerequisites

​1. Create the HTTP source table

​2. Create the Iceberg sink table

​3. Validate the pipeline query

​4. Create the pipeline

​5. Send an event

​6. Monitor the pipeline

​Continue

Prerequisites

1. Create the HTTP source table

2. Create the Iceberg sink table

3. Validate the pipeline query

4. Create the pipeline

5. Send an event

6. Monitor the pipeline

Continue