> ## Documentation Index
> Fetch the complete documentation index at: https://docs.e6data.com/llms.txt
> Use this file to discover all available pages before exploring further.

# HTTP to Iceberg

> Build a streaming pipeline that accepts JSON events over HTTP and writes them to an Apache Iceberg table.

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:

```bash theme={"theme":{"light":"github-light","dark":"github-dark"}}
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`:

```bash theme={"theme":{"light":"github-light","dark":"github-dark"}}
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`.

```bash theme={"theme":{"light":"github-light","dark":"github-dark"}}
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.

```bash theme={"theme":{"light":"github-light","dark":"github-dark"}}
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

```bash theme={"theme":{"light":"github-light","dark":"github-dark"}}
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.

```bash theme={"theme":{"light":"github-light","dark":"github-dark"}}
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:

```bash theme={"theme":{"light":"github-light","dark":"github-dark"}}
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:

```bash theme={"theme":{"light":"github-light","dark":"github-dark"}}
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](/ingest/examples/filesystem-to-iceberg) for an API-driven pipeline that reads files from object storage.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.