Load Precomputed Embeddings into Pinecone with a Daily Upsert Pipeline

Embedding jobs produce files; vector databases want upserts. This blueprint is the bridge for a RAG corpus. io.kestra.plugin.core.http.Download pulls a JSON export of precomputed embeddings, a small Python script reshapes each record into Pinecone's vector format and writes one map per line, and io.kestra.plugin.pinecone.Upsert loads the whole file through its from property in a single task. Because upserts overwrite by vector id, reruns refresh the corpus instead of duplicating it, and Slack gets the upserted count after every load.

How it works

1. download_embeddings (io.kestra.plugin.core.http.Download) fetches the JSON export named by the embeddings_url input into internal storage.
2. reshape_vectors (io.kestra.plugin.scripts.python.Script on the io.kestra.plugin.core.runner.Process task runner, no container needed) reads the export, builds one map per record with id, values, and any scalar fields as metadata, and writes them line by line to vectors.ion.
3. upsert_vectors (io.kestra.plugin.pinecone.Upsert) points its from property at the reshaped file and loads every vector in one call, outputting upsertedCount.
4. notify posts the upserted count, index name, and source URL to Slack.
5. The errors block posts a distinct Slack alert when any step fails, so a stale corpus never hides behind a green dashboard.
6. A disabled-by-default Schedule trigger refreshes the index daily at 05:00.

What you get

• A rerun-safe ingestion path: upserts key on vector id, so executing twice never duplicates a vector.
• A clean seam between embedding computation and vector loading; the flow consumes any JSON export with an id and an embedding per record.
• The reshape logic in twenty lines of Python you can adapt to any export schema.
• The upserted count as a first-class output, ready for alerting or reconciliation against the source.

Who it's for

• ML engineers who batch-compute embeddings upstream and need a reliable load into Pinecone.
• Data engineers treating the vector index as one more destination in their ELT landscape.
• RAG teams who want corpus refreshes scheduled, observable, and rerun-safe.

Why orchestrate this with Kestra

A load script on cron fails silently, keeps no history, and loses the file when the disk rotates. Kestra runs the same three steps with internal storage between them, passes the reshaped file to the upsert by URI, surfaces upsertedCount into the notification, and records every refresh in the execution history. When the export schema changes, the failure is a red execution with logs, not a quietly shrinking index.

Prerequisites

• An existing Pinecone index whose dimension matches the embeddings; the index bootstrap blueprint creates one.
• An HTTP-reachable JSON export of embedding records, each with id and embedding fields.
• Python 3 available on the Kestra host, since the script runs on the Process task runner.
• A Slack incoming webhook for load reports.

Secrets

• PINECONE_API_KEY: Pinecone API key with write permission on the index.
• SLACK_WEBHOOK_URL: Slack incoming webhook URL.

Quick start

1. Add the PINECONE_API_KEY and SLACK_WEBHOOK_URL secrets to your Kestra namespace.
2. Execute the flow with embeddings_url pointing at your embedding job's JSON export.
3. Check the Slack message reports the expected vector count, then set disabled: false on the daily trigger.

How to extend

• Add a namespace property on the upsert task to keep multiple corpora separated inside one index.
• Swap the download for an object storage task when the export lands in S3 or GCS instead of behind HTTP.
• Compare upsertedCount against the record count printed by the reshape step and fail the flow on a mismatch.
• Chain the semantic query report blueprint after the load as a retrieval smoke test.

Links

• Pinecone plugin documentation
• Pinecone documentation