---
title: "How do I replace an Airflow DAG and a vector database with one schema?"
description: "Replace an Airflow DAG, vector upsert, and API with typed columns, an embedding index, and HTTP routes in one file. Keep Airflow for unrelated batch jobs."
url: "https://www.pixeltable.com/learn/how-to/migrate-from-airflow-and-a-vector-database"
updated: "2026-10-28"
doc: "https://docs.pixeltable.com/howto/coming-from"
---

# How do I replace an Airflow DAG and a vector database with one schema?

An Airflow DAG plus a vector database plus an API is three systems for one retrieval job: extract, embed, upsert, then serve. The table that holds the files can also run the model and the index, and the same file can expose HTTP. Keep Airflow for a fleet of unrelated batch jobs.

Updated: 2026-10-28
Part of [How do I ... in Python?](https://www.pixeltable.com/learn/how-to).

## Before you start

About 25 min.

- Python 3.11+
- pip install -U 'pixeltable[serve]'
- pxt init in the project root
- sentence-transformers for all-MiniLM-L6-v2

## On this page

- [Before you start](https://www.pixeltable.com/learn/how-to/migrate-from-airflow-and-a-vector-database#before-you-start)
- [Ingest with typed columns instead of a drop folder](https://www.pixeltable.com/learn/how-to/migrate-from-airflow-and-a-vector-database#step-1)
- [Chunk and index in a view, not an upsert task](https://www.pixeltable.com/learn/how-to/migrate-from-airflow-and-a-vector-database#step-2)
- [Rank from the same file](https://www.pixeltable.com/learn/how-to/migrate-from-airflow-and-a-vector-database#step-3)
- [Serve insert and search over HTTP](https://www.pixeltable.com/learn/how-to/migrate-from-airflow-and-a-vector-database#step-4)
- [Comparison](https://www.pixeltable.com/learn/how-to/migrate-from-airflow-and-a-vector-database#comparison)
- [Where Pixeltable fits](https://www.pixeltable.com/learn/how-to/migrate-from-airflow-and-a-vector-database#where-pixeltable-fits)
- [Questions](https://www.pixeltable.com/learn/how-to/migrate-from-airflow-and-a-vector-database#questions)

## 1. Ingest with typed columns instead of a drop folder {#step-1}

The extract task in a DAG usually copies files into object storage, then a later task reads them. Declare Document or String columns so later transforms know the type. Insert is the ingest.

```python
import pixeltable as pxt
from pixeltable.functions.document import document_splitter
from pixeltable.functions.huggingface import sentence_transformer
from pixeltable.serving import FastAPIRouter

TableModel = pxt.model_base()
text_embed = sentence_transformer.using(model_id='all-MiniLM-L6-v2')

class Docs(TableModel, name='docs'):
    source: pxt.String
    document: pxt.Document
```

## 2. Chunk and index in a view, not an upsert task {#step-2}

The embed-and-upsert task writes vectors to Pinecone or pgvector. Here a splitter view emits passage rows and an embedding index follows each insert. There is no second store to keep in sync.

```python
class Chunks(
    TableModel,
    name='chunks',
    base=Docs,
    iterator=document_splitter(
        document=Docs.document, separators='sentence,token_limit', limit=300, overlap=50
    ),
):
    __indexes__ = [
        pxt.EmbeddingIndex(text, string_embed=text_embed),
    ]
```

## 3. Rank from the same file {#step-3}

The last DAG edge is usually a FastAPI service that queries the vector database. @pxt.query ranks chunks. You do not schedule a sensor that waits for the upsert to finish.

```python
@pxt.query
def search_docs(query: str, limit: int = 5):
    sim = Chunks.text.similarity(string=query)
    return (
        Chunks.order_by(sim, asc=False)
        .limit(limit)
        .select(Chunks.text, sim)
    )
```

## 4. Serve insert and search over HTTP {#step-4}

FastAPIRouter exposes the insert and the query. Apply the schema, then start the service. New files become searchable at insert. Keep Airflow if the work is many unrelated batch jobs.

```python
api = FastAPIRouter(name='docs_search')
api.add_insert_route(Docs, path='/docs', inputs=[Docs.source], uploadfile_inputs=[Docs.document])
api.add_query_route(path='/search', query=search_docs)

# pxt init
# pxt schema update app.py docs_search -f
# pxt service update app.py docs_search -f
```

## Airflow plus a vector database versus one schema {#comparison}

| Job | DAG plus vector DB plus API | One schema |
| --- | --- | --- |
| Ingest | An extract task and a drop folder | Insert into typed columns |
| Embed | A worker that calls a model | EmbeddingIndex on the chunk view |
| Upsert | A Pinecone or pgvector client | The index updates with the row |
| Serve | FastAPI you write against that store | FastAPIRouter insert and search |
| New file | Rerun the DAG or wait for a sensor | Insert that row |
| Keep it when | A fleet of unrelated batch jobs | One media and retrieval pipeline |

## Where Pixeltable fits {#where-pixeltable-fits}

Pixeltable vs Airflow is that split: the DAG is the schema when the unit of work is a row of media, models, and retrieval. Keep Airflow for unrelated batch jobs. Pinecone or pgvector stay only when that engine is the serving contract. Coming from another stack is the docs hub.

```python
import pixeltable as pxt
from pixeltable.functions.document import document_splitter
from pixeltable.functions.huggingface import sentence_transformer
from pixeltable.serving import FastAPIRouter

TableModel = pxt.model_base()
text_embed = sentence_transformer.using(model_id='all-MiniLM-L6-v2')

class Docs(TableModel, name='docs'):
    source: pxt.String
    document: pxt.Document

class Chunks(
    TableModel,
    name='chunks',
    base=Docs,
    iterator=document_splitter(
        document=Docs.document, separators='sentence,token_limit', limit=300, overlap=50
    ),
):
    __indexes__ = [
        pxt.EmbeddingIndex(text, string_embed=text_embed),
    ]

@pxt.query
def search_docs(query: str, limit: int = 5):
    sim = Chunks.text.similarity(string=query)
    return Chunks.order_by(sim, asc=False).limit(limit).select(Chunks.text, sim)

api = FastAPIRouter(name='docs_search')
api.add_insert_route(Docs, path='/docs', inputs=[Docs.source], uploadfile_inputs=[Docs.document])
api.add_query_route(path='/search', query=search_docs)

# pxt init
# pxt schema update app.py docs_search -f
# pxt service update app.py docs_search -f
```

## Concepts

- [What is a declarative pipeline?](https://www.pixeltable.com/learn/what-is-a-declarative-pipeline)
- [What is a computed column?](https://www.pixeltable.com/learn/what-is-a-computed-column)
- [How does an embedding index differ from a vector database?](https://www.pixeltable.com/learn/embedding-index-vs-vector-database)

## Use case

- [Declarative AI Infrastructure: Define Pipelines, Not Plumbing](https://www.pixeltable.com/use-cases/declarative-ai-data-infrastructure)

## Questions {#questions}

### Is this a replacement for every Airflow DAG? {#faq-1}

No. For media, models, and retrieval on one table, the DAG is the schema. For a fleet of unrelated batch jobs, keep Airflow. Pixeltable vs Airflow is that split.

### Do I still need Pinecone or pgvector? {#faq-2}

Only when that engine is already the serving contract. This schema indexes chunks on insert. Pixeltable vs Pinecone covers the hosted ANN case.

### What happens when a new file arrives? {#faq-3}

Insert a row. The splitter view and the embedding index run for that document. You do not rerun the DAG or wait for a sensor.

## Related

- [What is a declarative pipeline?](https://www.pixeltable.com/learn/what-is-a-declarative-pipeline)
- [What is a computed column?](https://www.pixeltable.com/learn/what-is-a-computed-column)
- [Embedding index vs a vector database](https://www.pixeltable.com/learn/embedding-index-vs-vector-database)
- [How do I build an AI data pipeline in Python?](https://www.pixeltable.com/learn/how-to/build-an-ai-data-pipeline)
- [How do I chunk PDFs for RAG in Python?](https://www.pixeltable.com/learn/how-to/chunk-pdfs-for-rag)
- [How do I migrate from Pinecone?](https://www.pixeltable.com/learn/how-to/migrate-from-pinecone)
- [Pixeltable vs Airflow](https://www.pixeltable.com/compare/pixeltable-vs-airflow)
- [Pixeltable vs Pinecone](https://www.pixeltable.com/compare/pixeltable-vs-pinecone)
- [Declarative data infrastructure](https://www.pixeltable.com/use-cases/declarative-ai-data-infrastructure)
- [Migrate to Pixeltable](https://docs.pixeltable.com/howto/coming-from)