Ingest an HTTP JSON Dataset into OpenSearch with Load

The fastest way to make a dataset searchable is three tasks. This blueprint downloads a public JSON array over HTTP with io.kestra.plugin.core.http.Download, reshapes it into one JSON object per line with a ten-line Python script, and hands the file to io.kestra.plugin.opensearch.Load, which bulk indexes it with each record's id as the document id, so reruns overwrite documents instead of duplicating them.

How it works

1. download_dataset (io.kestra.plugin.core.http.Download) fetches the JSON array and stores it in internal storage, exposed as {{ outputs.download_dataset.uri }}.
2. to_ndjson (io.kestra.plugin.scripts.python.Script on the Process task runner, no container needed) maps the file in through inputFiles, writes one JSON object per line to users.ndjson declared in outputFiles, and emits the record count through Kestra's output protocol as {{ outputs.to_ndjson.vars.record_count }}.
3. index_users (io.kestra.plugin.opensearch.Load) reads the file from {{ outputs.to_ndjson.outputFiles['users.ndjson'] }} and bulk indexes it into the users index. With idKey: id and removeIdKey: false, the record's own id becomes the document id and stays in the source. The task reports {{ outputs.index_users.size }} records sent.
4. notify posts both counts to Slack, converted and sent, so a mismatch surfaces immediately.
5. The errors block posts a distinct Slack alert naming the failed execution.

What you get

• A working HTTP to OpenSearch pipeline with no ingestion framework and no custom client code.
• Rerun-safe indexing: stable document ids mean executing twice never duplicates a record.
• Two independent counts, converted versus sent, acting as a built-in reconciliation check.
• File handoffs through internal storage, so each step is replayable in isolation.

Who it's for

• Search engineers seeding OpenSearch indexes from REST APIs or published datasets.
• Data engineers who want document ingestion in the same orchestrator as the rest of the stack.
• Anyone prototyping an OpenSearch-backed feature who needs real data indexed in minutes.

Why orchestrate this with Kestra

A one-off curl-python-bulk script does this once; a flow does it observably forever. Each stage's artifact lands in internal storage, so a failed load reruns without re-downloading, the counts land in Slack and in the execution history, and swapping the source URL or the target index is a one-line change reviewed in version control. Add a schedule and the same three tasks become a refresh pipeline.

Prerequisites

• A reachable OpenSearch cluster and credentials allowed to write to the users index.
• A Python 3 interpreter on the Kestra host, since the script runs on the Process task runner; only the standard library is used.
• A Slack incoming webhook for reports and alerts.

Secrets

• OPENSEARCH_HOST: OpenSearch endpoint, for example https://opensearch.example.com:9200.
• OPENSEARCH_USERNAME: OpenSearch username.
• OPENSEARCH_PASSWORD: OpenSearch password.
• SLACK_WEBHOOK_URL: Slack incoming webhook URL.

Quick start

1. Add the four secrets to your Kestra namespace.
2. Execute the flow, the demo endpoint needs no credentials, and check Slack reports ten records converted and sent.
3. Query the users index in OpenSearch Dashboards to see the documents.
4. Point uri at your real data source and rename the index.

How to extend

• Add a Schedule trigger to turn the one-shot ingestion into a recurring refresh.
• Enrich or filter records inside the Python script before they are written out, dropping fields or adding a load timestamp.
• Set opType: CREATE on the Load task to fail on duplicates instead of overwriting when appending immutable events.
• Verify the load with a follow-up io.kestra.plugin.opensearch.Search counting documents in the target index.

Links

• OpenSearch plugin documentation
• OpenSearch bulk indexing documentation