# Cocoindex Use Cases in Software Development: Declarative Data Pipelines for ETL, ML, and Live Streaming

> Explore Cocoindex use cases in software development. Build declarative, change-aware data pipelines for ETL, ML, and live streaming efficiently with Cocoindex.

- Repository: [CocoIndex/cocoindex](https://github.com/cocoindex-io/cocoindex)
- Tags: use-cases
- Published: 2026-05-05

---

**Cocoindex is a declarative, change-aware execution engine that enables software developers to build incremental data pipelines for ETL, document processing, ML feature extraction, and live streaming without manually managing state or change detection logic.**

The `cocoindex-io/cocoindex` repository provides a Python framework backed by a Rust core that treats data pipelines as declarative descriptions of target states rather than imperative scripts. By combining **memoization**, **stable component paths**, and **atomic state diffs**, Cocoindex eliminates boilerplate code for tracking changes and synchronizing external resources, making it particularly effective for modern software development workflows.

## Core Architecture: App, Component, and Target State

Cocoindex organizes code into three interlocking abstractions that enable declarative execution:

- **App** – The top-level runnable unit defined in `cocoindex._internal.app.App` that owns a root component and orchestrates pipeline updates.
- **Component** – A processing unit created via `mount()` or `use_mount()` in `cocoindex._internal.api.mount`, each owning a stable **component path** that persists across runs.
- **Target State** – A declarative description of external resources (files, database rows, vector store entries) defined in `cocoindex._internal.target_state.TargetState`, which the engine diffs against previous runs to apply only necessary creates, updates, or deletes.

When a pipeline runs, the engine computes a **stable path** (`StablePath`) for each component and fingerprints the logic using AST-based analysis. This allows Cocoindex to determine exactly which components changed and require re-execution.

## Declarative → Incremental → Atomic Execution

Cocoindex pipelines follow a three-phase execution model implemented across the Python layer and the Rust core:

1. **Declarative** – Developers use high-level APIs like `coco.mount`, `coco.use_mount`, and `coco.mount_target` to declare what the final data state should look like, eliminating explicit "write" or "update" logic.
2. **Incremental** – The engine records fingerprints of function logic and input data via `cocoindex._internal.function.SyncFunction.__call__` and `_compute_logic_fingerprint`. Only components with changed inputs or logic are re-executed.
3. **Atomic** – The **Rust core** (`rust/core`) applies each component's target-state diff in a single transaction. If a component disappears, its target states are automatically cleaned up.

This architecture ensures that pipelines are deterministic, efficient, and safe to rerun without side effects.

## Top Cocoindex Use Cases in Software Development

### ETL and Document Processing Pipelines

Cocoindex excels at extracting, transforming, and loading file-based data. The framework's `localfs` connector, combined with memoized functions, creates efficient document processing workflows.

The `files_transform` example demonstrates converting Markdown files to HTML incrementally:

```python
import pathlib
import cocoindex as coco
from cocoindex.resources.file import FileLike, PatternFilePathMatcher
from cocoindex.connectors import localfs
from markdown_it import MarkdownIt

_md = MarkdownIt("gfm-like")

@coco.fn(memo=True)
async def process_file(file: FileLike, outdir: pathlib.Path) -> None:
    html = _md.render(await file.read_text())
    outname = "__".join(file.file_path.path.parts) + ".html"
    localfs.declare_file(outdir / outname, html, create_parent_dirs=True)

@coco.fn
async def app_main(sourcedir: pathlib.Path, outdir: pathlib.Path) -> None:
    files = localfs.walk_dir(
        sourcedir,
        path_matcher=PatternFilePathMatcher(included_patterns=["**/*.md"]),
        live=True,
    )
    await coco.mount_each(process_file, files.items(), outdir)

app = coco.App(
    coco.AppConfig(name="FilesTransform"),
    app_main,
    sourcedir=pathlib.Path("./data"),
    outdir=pathlib.Path("./output_html"),
)

```

**Key implementation details:**
- `@coco.fn(memo=True)` deduplicates work on unchanged files by storing fingerprints in [`cocoindex._internal.memo_fingerprint.py`](https://github.com/cocoindex-io/cocoindex/blob/main/cocoindex._internal.memo_fingerprint.py).
- `mount_each` creates a component per Markdown file, each with its own stable path.
- The `localfs` connector declares output files as target states; only new or changed files are written.

Source: [[`examples/files_transform/main.py`](https://github.com/cocoindex-io/cocoindex/blob/main/examples/files_transform/main.py)](https://github.com/cocoindex-io/cocoindex/blob/main/examples/files_transform/main.py)

### ML Feature Extraction and Vector Search

For machine learning workflows, Cocoindex provides automatic memoization of expensive model calls and native connectors for vector databases like Qdrant.

```python
import pathlib
import cocoindex as coco
from cocoindex.connectors import qdrant
from cocoindex.ops.sentence_transformers import SentenceTransformerEmbedder

embedder = SentenceTransformerEmbedder(model_name="all-MiniLM-L6-v2")

@coco.fn(memo=True)
async def embed_file(file: FileLike, target: qdrant.CollectionTarget) -> None:
    txt = await file.read_text()
    vec = embedder.embed(txt)
    target.declare_vector(
        id=file.file_path.path.as_posix(),
        vector=vec,
        payload={"source": str(file.file_path)},
    )

@coco.fn
async def app_main(sourcedir: pathlib.Path, collection: str) -> None:
    files = localfs.walk_dir(sourcedir, live=True)
    target = await coco.use_mount(qdrant.declare_collection_target, collection)
    await coco.mount_each(embed_file, files.items(), target)

coco.App("DocsEmbedding", app_main,
          sourcedir=pathlib.Path("./docs"),
          collection="my_docs").update_blocking(report_to_stdout=True)

```

The **SentenceTransformerEmbedder** runs only when inputs change, and embeddings are stored declaratively via `declare_vector`. The Qdrant connector handles the complexity of upserting vectors only when the source document content changes.

Source: [[`examples/text_embedding_qdrant/main.py`](https://github.com/cocoindex-io/cocoindex/blob/main/examples/text_embedding_qdrant/main.py)](https://github.com/cocoindex-io/cocoindex/blob/main/examples/text_embedding_qdrant/main.py)

### Live Streaming and Event Processing

Cocoindex supports continuous data sources through **LiveComponent**, which runs in the background and processes incremental updates without restarting the pipeline.

```python
import cocoindex as coco
from cocoindex.connectors import kafka, lancedb
from cocoindex.ops import litellm

class KafkaToLance(coco.LiveComponent):
    def __init__(self, topic: str, collection: str):
        self.topic = topic
        self.collection = collection

    async def process_live(self, operator: coco.LiveComponentOperator) -> None:
        async for msg in kafka.consume(self.topic):
            summary = await litellm.summarize(msg.value)
            await operator.target.declare_row(id=msg.offset, payload=summary)

async def app_main(topic: str, collection: str) -> None:
    target = await coco.use_mount(lancedb.declare_collection_target, collection)
    await coco.mount(KafkaToLance(topic, collection), target)

coco.App("KafkaLance", app_main, topic="events", collection="events").update()

```

Live components are instantiated via `coco.mount` when `is_live_component_class` triggers the `_mount_live_component` path in [`cocoindex._internal.api.py`](https://github.com/cocoindex-io/cocoindex/blob/main/cocoindex._internal.api.py). This enables use cases like Kafka-to-Database streaming, real-time log processing, and continuous file system monitoring.

Source: [[`examples/kafka_to_lancedb/main.py`](https://github.com/cocoindex-io/cocoindex/blob/main/examples/kafka_to_lancedb/main.py)](https://github.com/cocoindex-io/cocoindex/blob/main/examples/kafka_to_lancedb/main.py)

## Context-Based Dependency Injection

Cocoindex replaces global singletons with **typed context keys** (`ContextKey`) provided during the environment lifespan. The current component's context is stored in a `ContextVar`, making dependencies automatically available downstream.

Define a key:

```python
PG_DB = coco.ContextKey[postgres.PgDatabase]("pg_db")

```

Provide it in the lifespan:

```python
with coco.environment.builder() as env:
    env.provide(PG_DB, pg_pool)

```

Consume it inside a component:

```python
db = coco.use_context(PG_DB)  # resolved from ComponentContext

```

Implementation details reside in `cocoindex._internal.component_ctx` and `cocoindex._internal.api`.

## Extensible Connectors and Performance Optimization

Cocoindex ships with connectors for PostgreSQL, Qdrant, LanceDB, Google Drive, and local filesystems. Each connector implements a `declare_*_target` function returning a `TargetStateProvider`, making it straightforward to add new storage backends.

For performance:

- **Runner abstraction** – Offload CPU-heavy functions to thread or GPU pools using `coco.runner.GPU` or `coco.runner.Runner`.
- **Automatic batching** – Decorate async functions with `@coco.fn(batching=True)` to process multiple inputs together. The batching logic lives in `cocoindex._internal.function.AsyncFunction`.

## Summary

- Cocoindex provides a **declarative execution model** where developers describe target states rather than writing imperative update logic.
- The engine automatically handles **incremental updates** through stable component paths, memoization, and logic fingerprinting in `cocoindex._internal.function` and `cocoindex._internal.stable_path`.
- **LiveComponent** enables continuous processing of streaming data sources like Kafka without pipeline restarts.
- **Context-based dependency injection** replaces global state with typed keys provided via `coco.environment.builder()`.
- The **Rust core** guarantees atomic application of state diffs and automatic cleanup of removed components.
- Built-in connectors for vector stores, databases, and filesystems make Cocoindex suitable for ETL, ML feature extraction, and real-time data synchronization.

## Frequently Asked Questions

### What is Cocoindex used for in software development?

Cocoindex is used to build **declarative data pipelines** that synchronize data between sources and destinations. Common use cases include ETL workflows (transforming Markdown to HTML), ML feature extraction (generating and caching text embeddings), incremental builds (processing only changed source files), and live streaming (moving data from Kafka to vector databases). The framework handles change detection, memoization, and atomic updates automatically.

### How does Cocoindex handle incremental updates?

Cocoindex computes **stable paths** and **fingerprints** for every component. When you decorate a function with `@coco.fn(memo=True)`, the engine stores a hash of the function's AST logic and input arguments in [`cocoindex._internal.memo_fingerprint.py`](https://github.com/cocoindex-io/cocoindex/blob/main/cocoindex._internal.memo_fingerprint.py). On subsequent runs, only components with changed fingerprints are re-executed, making pipelines efficient even with large datasets.

### What is the difference between `mount()` and `mount_each()`?

`mount()` creates a single component from a class or function, suitable for singleton resources or live components. `mount_each()` creates multiple components from an iterable, generating a unique component with its own stable path for each item. Use `mount_each()` when processing collections like directories of files or database result sets, as shown in the text embedding and file transform examples.

### Can Cocoindex process live streaming data?

Yes. Cocoindex provides **LiveComponent**, a subclass that runs continuously in the background. When you subclass `coco.LiveComponent` and implement `process_live()`, the runtime keeps the component alive after the initial `mark_ready` call. This enables processing of streaming sources like Kafka topics, file system watches, or message queues without tearing down and rebuilding the entire pipeline.