> For the complete documentation index, see [llms.txt](https://docs.e6data.com/ingestion-engine/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.e6data.com/ingestion-engine/get-started/examples/http-to-iceberg.md). # 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 e6 Ingestion 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 e6 Ingestion Engine API; * an e6 Ingestion Engine API bearer token; * an Iceberg REST catalog; and * an S3 or S3-compatible warehouse accessible to the e6 Ingestion Engine. Set the API values used by the requests: ```bash export INGESTION_ENGINE_API_URL="https://" export INGESTION_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 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": "" } }, "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 e6 Ingestion 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 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": "", "warehouse": "s3:///warehouse", "auth": { "token": "" } }, "namespace": "webhooks", "tableName": "events", "appendOnly": true, "partitioning": { "fields": [ { "name": "event_time", "transform": "hour" } ], "shuffle_by_partition": { "enabled": true } }, "storageOptions": { "s3.endpoint": "", "s3.region": "", "s3.access-key-id": "", "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 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 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 e6 Ingestion Engine environment. This URL is separate from `INGESTION_ENGINE_API_URL` and should not be an internal service address. ```bash export HTTP_SOURCE_URL="https://" export 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 curl --fail-with-body --silent --show-error \ "$INGESTION_ENGINE_API_URL/api/v1/pipelines/" \ -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 curl --fail-with-body --silent --show-error \ "$INGESTION_ENGINE_API_URL/api/v1/pipelines//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](/ingestion-engine/get-started/examples/filesystem-to-iceberg.md) for an API-driven pipeline that reads files from object storage. --- # Agent Instructions This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com. ## Querying This Documentation If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question. Perform an HTTP GET request on the current page URL with the `ask` query parameter, and the optional `goal` query parameter: ``` GET https://docs.e6data.com/ingestion-engine/get-started/examples/http-to-iceberg.md?ask=&goal= ``` `ask` is the immediate question: it should be specific, self-contained, and written in natural language. `goal` is optional and describes the broader end goal you are ultimately trying to accomplish on behalf of the user. GitBook uses it to tailor the answer towards what is most useful for that goal. The response will contain a direct answer to the question and relevant excerpts and sources from the documentation. Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.