How to

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?

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

1. Ingest with typed columns instead of a drop folder

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.

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

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.

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

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.

@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

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.

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

Airflow plus a vector database versus one schema
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

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.

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

Questions

Is this a replacement for every Airflow DAG?

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?

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?

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.

Next step

Copy the commands into an agent, or run them locally with pip install pixeltable.

Get started Quick Start