Extract data from a CSV file, load it in batch to Weaviate and query it

Build a reproducible CSV-to-Weaviate ingestion pipeline that downloads a remote CSV file, converts it to Kestra's internal ION format, batch-creates objects in a Weaviate vector database class, and validates the load with a GraphQL query. This blueprint solves the common problem of getting structured tabular data into a vector store quickly so you can power semantic search, retrieval-augmented generation (RAG), and AI applications without hand-rolling glue scripts.

How it works

The flow runs four sequential tasks against a Weaviate cluster:

1. csv (io.kestra.plugin.core.http.Download) fetches the trivia questions CSV from a public Hugging Face dataset URL.
2. csv_to_ion (io.kestra.plugin.serdes.csv.CsvToIon) converts the CSV stream to ION, the internal format Kestra plugins consume row by row.
3. batch_load (io.kestra.plugin.weaviate.BatchCreate) inserts every row into the QuestionsCsv class on your Weaviate cluster, using the url and apiKey variables.
4. query (io.kestra.plugin.weaviate.Query) runs a GraphQL Get query that returns 10 objects with the answer, category, and question fields to confirm the load succeeded.

Cluster connection details live in variables.host and variables.secret, so you can repoint the flow to staging or production by editing two lines.

What you get

• A working CSV to vector database ingestion pattern you can adapt to any tabular source.
• Batch object creation that handles arbitrary row counts via the streaming ION format.
• A built-in GraphQL smoke test that proves data landed in the right class.
• A declarative YAML pipeline that runs the same way locally, in CI, and in production.

Who it's for

• AI and ML engineers building RAG pipelines on top of Weaviate.
• Data engineers loading reference datasets, product catalogs, or knowledge bases into a vector store.
• Platform teams standardizing how application data reaches Weaviate clusters.

Why orchestrate this with Kestra

Weaviate has no native scheduler or ingestion runner: client SDKs assume you already have one. Kestra fills that gap with event and schedule triggers, automatic retries on transient HTTP and API errors, full input and output lineage between csv, csv_to_ion, batch_load, and query, and declarative YAML you can review in pull requests. The same flow runs from a webhook, a cron schedule, or an upstream flow finishing.

Prerequisites

• A running Weaviate cluster (Weaviate Cloud or self-hosted) reachable from your Kestra workers.
• An API key with permission to create classes and insert objects.
• Network access to the source CSV URL (or substitute your own).

Secrets

The flow currently reads the API key from variables.secret as a placeholder. For production, replace it with a Kestra secret, for example:

• WEAVIATE_API_KEY: API key for the target Weaviate cluster, referenced as {{ secret('WEAVIATE_API_KEY') }}.

Quick start

1. Copy the flow into your Kestra instance under a namespace you control.
2. Update variables.host to point at your Weaviate cluster URL.
3. Store your API key as a secret and replace variables.secret with {{ secret('WEAVIATE_API_KEY') }}.
4. Execute the flow and inspect the query task output to confirm 10 objects are returned.

How to extend

• Swap the csv download for io.kestra.plugin.aws.s3.Download, io.kestra.plugin.gcp.gcs.Download, or any other source plugin to ingest from object storage.
• Add a Schedule trigger to refresh the Weaviate class on a cadence, or a Webhook trigger to load on demand.
• Replace BatchCreate with delete-then-create logic to support full refreshes, or add upserts keyed by a stable ID field.
• Chain a downstream task that calls an LLM (OpenAI, Anthropic, Vertex AI) using the queried objects as RAG context.
• Parameterize className, limit, and the CSV URL as flow inputs so the same blueprint serves multiple datasets.

Links

• plugin-weaviate documentation