> 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/filesystem-to-iceberg.md). # Filesystem to Iceberg This example reads newline-delimited JSON files from Amazon S3, transforms each record with SQL, and writes the result to an Apache Iceberg table registered in AWS Glue Data Catalog. 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 e6 Ingestion 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: ```bash export INGESTION_ENGINE_API_URL="https://" export INGESTION_ENGINE_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](/ingestion-engine/connectors/storage/filesystem.md) and [Iceberg](/ingestion-engine/connectors/lakehouse/iceberg.md) references. ## 1. Create the Filesystem Source Table Create a typed Filesystem table through `POST /api/v1/connectors/filesystem/tables`: ```bash 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:///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`: ```bash 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" }, "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: ```bash 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`: ```bash 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: ```bash curl --fail-with-body \ "$INGESTION_ENGINE_API_URL/api/v1/pipelines/" \ -H "Authorization: Bearer $INGESTION_ENGINE_API_TOKEN" curl --fail-with-body \ "$INGESTION_ENGINE_API_URL/api/v1/pipelines//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](/ingestion-engine/architecture/overview.md) to see how the controller, workers, connectors, and checkpoints execute these pipelines. --- # 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/filesystem-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.