Lance

The Lance plugin adds a lance format to flyte.io.DataFrame, so a Lance dataset can be passed between tasks as a typed input and output.

Lance datasets are lazy. Opening one reads its manifest and nothing else; rows arrive when you ask for them, sequentially for a scan or by index for a shuffled training loop. The plugin is built to keep that intact across a task boundary, so the decoder hands you a live lance.LanceDataset rather than a table it downloaded on your behalf.

Lance vs. Parquet

Both are columnar and Arrow-native. Read a dataset end to end and they cost about the same. The difference shows up when you want a subset of the rows.

Parquet stores rows in row groups, and a row group is the smallest thing you can decode. Ask for fifty scattered rows and you pay for every row group they happen to land in. Flyte’s built-in Parquet decoder is blunter still: it materializes the entire table into memory, and only then can you index into it.

Lance addresses rows individually, and its vector and scalar indices let a query jump to the matching ones. Fifty scattered rows costs about fifty rows.

Which one should you use?

Parquet, if your access pattern is “read all of it” or “read a whole partition of it”. That covers most ETL, aggregation, and anything headed for a SQL engine. It is also what you want when the data is going to a team whose tooling you don’t control, because everything reads Parquet and that is not a small advantage.

Lance, if you go after scattered rows:

• Shuffled training: SGD wants a different random order every epoch. Lance serves it by random access; Parquet re-reads the dataset.
• Multimodal rows: Image or audio bytes stored beside their labels, where most tasks touch only the labels.
• Vector search: Embeddings with an ANN index in the same dataset as the data they describe.
• Point lookups: Fetching individual records by id from a large table.
• Datasets bigger than memory: Stream them rather than partition them by hand.

Both are formats on flyte.io.DataFrame, so one task can produce Parquet and another Lance in the same workflow. What you cannot do is move a dataset cheaply between them. A lance input arrives as a lance.LanceDataset or a pyarrow.Table, and there is no third option, so this is a choice you make per dataset rather than per task. Other dataframe types covers what that rules out.

How much does it matter?

The benchmark below pulls 1,000 random rows out of 100,000, each carrying a 512-byte payload. Parquet materializes a 53.6 MB table to answer that. Lance materializes just 0.54 MB, which is only the 1,000 rows you actually asked for.

The 100x gap is the more reliable number because it comes from how the data is laid out, not the machine it’s running on. Flyte’s Parquet decoder has to build the entire table before it can index into it, and a warm cache doesn’t change that. The actual wall-clock speedup is less dramatic: around 10x on local disk and 3.5x against object storage in the same benchmark. Those numbers will vary with row size and batch size though.

Installation

pip install flyteplugins-lance

Put the plugin in your task image and you’re done. Flyte finds it through the flyte.plugins.types entry point and registers the lance format on startup, so there is nothing to import and nothing to call:

import flyte

image = flyte.Image.from_debian_base(name="lance").with_pip_packages("flyteplugins-lance")

env = flyte.TaskEnvironment(
    name="lance_env",
    image=image,
    resources=flyte.Resources(cpu="1", memory="2Gi"),
)

What the plugin registers

Four handlers, against Flyte’s dataframe transformer engine:

lance.LanceDataset already defaults to the lance format, so you never have to annotate it. pyarrow.Table doesn’t, and stays on Parquet unless you say otherwise. That is the one place the format shows up in your type signatures.

Other dataframe types

Those four are the entire surface, which is worth knowing before you design a workflow around it. The Polars plugin can hand data to pandas and PySpark because all three speak Parquet. Lance has no such common currency here, so there is no lance-to-pandas or lance-to-Polars handler. So do the conversion yourself inside the task. Take the Arrow table and call to_pandas(), or take the handle and convert a batch at a time:

# `to_pandas()` is a pyarrow method, but it still needs pandas installed in the
# task image: .with_pip_packages("flyteplugins-lance", "pandas")

@env.task
async def as_pandas(table: pa.Table) -> int:
    """Convert inside the task. Declaring `pd.DataFrame` directly would fail:
    the plugin registers no lance-to-pandas handler."""
    df = table.to_pandas()
    return len(df)

@env.task
async def as_pandas_streaming(ds: lance.LanceDataset) -> int:
    """Convert per batch instead, so the dataset is never held whole."""
    rows = 0
    for batch in ds.scanner(columns=["id"], batch_size=1024).to_batches():
        rows += len(batch.to_pandas())
    return rows

On anything large, use the second one. A whole-table to_pandas() throws away the streaming property you picked Lance for, and hands the result to pandas, which will sit on it in memory.

Passing a dataset between tasks

Two ways to do this, and they are not interchangeable.

Return a live handle

Return a lance.LanceDataset, accept one on the other side. No wrapper, no open():

import tempfile

import lance
import pyarrow as pa

@env.task
async def build_dataset(n: int = 10_000) -> lance.LanceDataset:
    uri = f"{tempfile.mkdtemp()}/points.lance"
    lance.write_dataset(pa.table({"id": list(range(n)), "value": [i * i for i in range(n)]}), uri)
    return lance.dataset(uri)

@env.task
async def summarize(ds: lance.LanceDataset) -> dict:
    # `ds` is a live handle. Stream it in batches; nothing is materialized whole.
    total = 0
    for batch in ds.scanner(columns=["value"], batch_size=1024).to_batches():
        total += sum(batch.column("value").to_pylist())
    return {"rows": ds.count_rows(), "sum_of_values": total}

The ds the consumer gets is already open. count_rows() answers from metadata, to_batches() pulls a batch at a time, and at no point does the whole dataset come down.

Encoding is memory-bounded too, since it streams the source through an Arrow RecordBatchReader. Bounded is not the same as cheap, though: every fragment still gets read and written again.

Hand off by reference

If the task already wrote the dataset, usually in chunks because it was too big to hold, don’t make Flyte read it back. Hand over the path:

import os

@env.task
async def convert(n: int = 10_000, chunk: int = 2_000) -> DataFrame:
    """Write a Lance dataset in bounded chunks, then hand it off by reference."""
    uri = os.path.join(tempfile.mkdtemp(), "dataset.lance")
    mode = "create"
    for start in range(0, n, chunk):
        rows = list(range(start, min(start + chunk, n)))
        lance.write_dataset(pa.table({"id": rows}), uri, mode=mode)
        mode = "append"

    # Flyte uploads the .lance directory verbatim: no re-read, no re-encode.
    return DataFrame(uri=uri, format="lance")

@env.task
async def inspect(ds: lance.LanceDataset) -> dict:
    # The consumer still takes the raw Lance type; the "lance" format resolves
    # the handoff regardless of which form the producer returned.
    return {"rows": ds.count_rows(), "fragments": len(ds.get_fragments())}

DataFrame(uri=..., format="lance") uploads the .lance directory as it sits. No Arrow round trip. The consumer can still ask for a plain lance.LanceDataset; the lance format sorts out the handoff no matter which form the producer chose.

Choosing between them

The distinction is mechanical, not stylistic. Return a lance.LanceDataset and the encoder runs: it reads your dataset through Arrow and writes a new one at the destination. Return a DataFrame carrying a URI and no encoder runs at all, because Flyte just uploads the directory recursively, the same way it treats any pre-written path for any format.

So one gives you a copy of a directory and the other gives you a dataset reconstructed from its rows. Whatever isn’t in those rows doesn’t survive the trip:

Rough rule: the handle is fine when the task just built something small in memory. Reach for the reference once the dataset is large, was written in chunks, or carries indices or blob columns.

Neither form copies data at call time, which means the usual dataframe caching caveat applies here: a downstream cached task won’t hit on identical content sitting at a new path. Content-based caching is the fix.

Arrow tables and the explicit lance format

pyarrow.Table defaults to Parquet, so this is where you name the format explicitly. Annotated[DataFrame, "lance"] does it:

from collections import OrderedDict
from typing import Annotated

from flyte.io import DataFrame

@env.task
async def build_table() -> Annotated[DataFrame, "lance"]:
    # A bare `DataFrame` or `pa.Table` would be stored as Parquet. The "lance"
    # annotation is what selects this plugin's encoder.
    table = pa.table(
        {
            "city": ["NYC", "SF", "LA", "SEA"],
            "temp_c": [7, 15, 20, 11],
            "humidity": [55, 70, 40, 80],
        }
    )
    return DataFrame.wrap_df(table)

@env.task
async def read_as_dataset(ds: lance.LanceDataset) -> int:
    # The same stored bytes, opened lazily as a streaming handle.
    return ds.count_rows()

@env.task
async def read_as_table(table: Annotated[pa.Table, OrderedDict(city=str, temp_c=int)]) -> dict:
    # Decoded eagerly into memory, narrowed to the two annotated columns.
    return {"columns": table.column_names, "rows": table.num_rows}

Those same stored bytes read back either way. read_as_dataset gets a streaming handle. read_as_table decodes eagerly into memory, narrowed to whichever columns its OrderedDict annotation names.

Eager decode pulls in the whole dataset, which is fine for a small lookup table and wrong for anything big or multimodal.

Reading only what you need

Narrowing happens on the open handle, inside the task, and because the dataset is lazy it comes straight off the bytes you fetch:

@env.task
async def sample_rows(ds: lance.LanceDataset) -> dict:
    # Random access: read only these rows, only these columns.
    rows = ds.take([0, 500, 9_999], columns=["id", "value"]).to_pylist()

    # Predicate pushdown: the filter runs inside Lance, so unmatched rows are
    # never decoded or sent over the wire.
    matched = ds.scanner(columns=["id"], filter="value > 98000000").to_table().num_rows

    return {"sample": rows, "matched": matched}

• Column projection: columns=[...] on scanner() or take(). Columns you don’t name are never touched.
• Predicate pushdown: filter="value > 98000000" is SQL, evaluated inside Lance. Rows that don’t match are never decoded, and a scalar index on that column turns the whole scan into a lookup.
• Random access: take([...]) fetches the row indices you hand it. This is the one Parquet can’t do cheaply, and it is what makes shuffled training practical.

There’s a fourth, batch_size on scanner(), in the streaming example above. It caps peak memory no matter how big the dataset gets.

Multimodal data

Blob encoding moves large binary values out of the regular column layout, so a scan that doesn’t ask for the column doesn’t pay for it. Mark the field in the schema:

import os
import random
import tempfile

import flyte
import lance
import pyarrow as pa
from flyte.io import DataFrame

image = flyte.Image.from_debian_base(name="lance-multimodal").with_pip_packages("flyteplugins-lance")

env = flyte.TaskEnvironment(
    name="lance_multimodal",
    image=image,
    resources=flyte.Resources(cpu="2", memory="4Gi"),
)

# `lance-encoding:blob` keeps large values out of the regular column layout, so a
# scan that doesn't ask for `image` never pays for the image bytes.
SCHEMA = pa.schema(
    [
        pa.field("id", pa.int32()),
        pa.field("image", pa.large_binary(), metadata={"lance-encoding:blob": "true"}),
        pa.field("label", pa.int32()),
    ]
)

Write it in chunks and hand it off by reference, which blob columns require anyway:

@env.task
async def convert(n: int = 4_000, chunk: int = 512) -> DataFrame:
    """Fold many small samples into one Lance dataset, a chunk at a time."""
    uri = os.path.join(tempfile.mkdtemp(), "images.lance")

    mode = "create"
    for start in range(0, n, chunk):
        rows = list(range(start, min(start + chunk, n)))
        table = pa.table(
            {
                "id": rows,
                "image": [_fake_image_bytes(i) for i in rows],
                "label": [i % 10 for i in rows],
            },
            schema=SCHEMA,
        )
        lance.write_dataset(table, uri, mode=mode)
        mode = "append"

    # Hand off by reference. Returning a `lance.LanceDataset` here would re-encode
    # the dataset through Arrow, which a blob-encoded column does not survive.
    return DataFrame(uri=uri, format="lance")

Streaming a shuffled epoch

Draw a fresh random order each epoch, pull it in batches, straight out of object storage. Labels come from take() and the image bytes from take_blobs(), both scoped to the same rows:

@env.task
async def train_one_epoch(df: DataFrame, batch_size: int = 128, seed: int = 0) -> dict:
    """Stream one shuffled epoch by random access. Nothing is downloaded whole."""
    ds = await df.open(lance.LanceDataset).all()

    order = list(range(ds.count_rows()))
    random.Random(seed).shuffle(order)

    seen = 0
    image_bytes = 0
    label_counts: dict[int, int] = {}
    for i in range(0, len(order), batch_size):
        rows = order[i : i + batch_size]

        # Structured columns come back inline. A blob column does not: `take` and
        # `scanner` hand you a {position, size} descriptor rather than the value,
        # so the bytes are read through `take_blobs`, which opens each one as a
        # file object. Either way only these rows are touched, so memory stays
        # proportional to the batch and not to the dataset.
        labels = ds.take(rows, columns=["label"]).column("label").to_pylist()

        for blob, label in zip(ds.take_blobs("image", indices=rows), labels):
            with blob as f:
                image_bytes += len(f.readall())  # stand-in for decode + augment
            label_counts[label] = label_counts.get(label, 0) + 1
            seen += 1

    return {
        "rows_streamed": seen,
        "image_bytes_read": image_bytes,
        # Map keys are stringified so the run UI renders them as text.
        "labels": {str(k): v for k, v in sorted(label_counts.items())},
    }

Peak memory is one batch whether the dataset holds 4,000 rows or 40 million.

Tar and WebDataset stream this just as well, to be fair. What they can’t do is shuffle it properly: reaching an arbitrary row means reading forward to it, so they approximate with interleaved shards and a buffer window. Lance addresses rows directly, so the shuffle is the real thing.

Reading the labels without the images

The images being blob-encoded means a scan over the structured columns skips them:

@env.task
async def label_histogram(df: DataFrame) -> dict:
    """Scan the structured columns only. The image bytes are never read."""
    ds = await df.open(lance.LanceDataset).all()

    counts: dict[int, int] = {}
    for batch in ds.scanner(columns=["label"], batch_size=4_096).to_batches():
        for label in batch.column("label").to_pylist():
            counts[label] = counts.get(label, 0) + 1
    return {str(k): v for k, v in sorted(counts.items())}

A BlobFile is a file object, so you can also read part of a value rather than all of it, which is how you inspect headers or sample a few frames without pulling whole images across:

@env.task
async def inspect_large_images(df: DataFrame, top_k: int = 3) -> list[int]:
    """Open blobs as file-like objects instead of loading them into memory."""
    ds = await df.open(lance.LanceDataset).all()

    sizes = []
    for blob in ds.take_blobs("image", indices=list(range(top_k))):
        with blob as f:
            sizes.append(len(f.readall()))
    return sizes

Parallel reads across fragments

Lance stores a dataset as fragments, which makes them an obvious unit of parallelism: give each mapped task one fragment id and every worker reads a disjoint set of files.

Pass the dataset as a flyte.io.DataFrame here, not a lance.LanceDataset. The reference form leaves the bytes alone, so a fragment id the parent computed still points at the same fragment in every child. Return a handle instead and the re-encode coalesces the fragments underneath you, leaving those ids pointing at nothing:

from functools import partial

@env.task
async def scan_fragment(df: DataFrame, fragment_id: int) -> int:
    """Read exactly one fragment. Each worker touches a disjoint slice of files."""
    ds = await df.open(lance.LanceDataset).all()
    return ds.get_fragment(fragment_id).to_table(columns=["id"]).num_rows

@env.task
async def fan_out(df: DataFrame) -> int:
    # Pass the dataset as a `DataFrame`, not a `lance.LanceDataset`: the reference
    # form keeps the stored bytes (and therefore the fragment ids) identical for
    # every worker. Returning a `lance.LanceDataset` would re-encode the dataset
    # at the boundary and coalesce the fragments, invalidating these ids.
    ds = await df.open(lance.LanceDataset).all()
    fragment_ids = [f.fragment_id for f in ds.get_fragments()]

    total = 0
    async for count in flyte.map.aio(partial(scan_fragment, df), fragment_ids):
        if isinstance(count, Exception):
            raise count
        total += count
    return total

Every worker opens the same URI, so nothing gets copied per worker and how wide you fan out is a question about your cluster, not your storage.

Indices and schema evolution

These belong to Lance, not to the plugin. They show up here because the handoff form decides whether they make it across a task boundary.

Vector and scalar indices: Build them on the open handle: create_index() for vectors (IVF_PQ, IVF_HNSW_PQ, IVF_HNSW_SQ), create_scalar_index() for the rest (BTREE, BITMAP, INVERTED, and others). Query with scanner(nearest={"column": "vec", "q": query, "k": 10}). A scalar index also speeds up the predicate pushdown above, including as a pre-filter on a vector search. Because indices live inside the dataset directory, they ride along on a reference handoff and vanish on a handle return.

Schema evolution: add_columns() adds a column without rewriting the ones already on disk, and can compute it from a SQL expression over the existing ones. It is cheap, and it is worth reaching for on a dataset your own task just built, before you hand it on.

Apply it to an input, though, and it writes to the producing task’s output. The decoder hands you a live handle onto storage rather than a copy, which is what makes streaming work, and Lance’s in-place operations write wherever the dataset actually lives. Treat a dataset you were handed as read-only and return a new one instead.

Lance’s versioning comes out of that same in-place model, and there is not much for it to do here: Flyte writes every output once, to a new path, so the only versions you will ever see are the producing task’s own write steps.

Object storage and credentials

The plugin feeds Flyte’s storage configuration into Lance’s storage_options, so reads and writes against remote storage use the same credentials as the rest of the run:

In practice this means a task on a cluster picks up its IAM role or workload identity without you doing anything, and the code you ran locally is the code that runs there. Lance pulls only the row ranges and columns each batch needs, direct from the bucket, with no staging step on local disk.

One wrinkle worth knowing if you ever assemble storage_options by hand: Lance spells the S3 endpoint key aws_endpoint, where several other object-store-backed libraries use aws_endpoint_url. The plugin already accounts for it.

Measuring the difference

The benchmark writes the same data twice, once as Parquet and once as Lance, then fetches an identical shuffled batch from each and renders a report. Both producers build the same Arrow table:

import random
import time
from typing import Annotated

import flyte
import flyte.report
import lance
import pyarrow as pa
from flyte.io import DataFrame

image = flyte.Image.from_debian_base(name="lance-benchmark").with_pip_packages("flyteplugins-lance")

env = flyte.TaskEnvironment(
    name="lance_benchmark",
    image=image,
    resources=flyte.Resources(cpu="2", memory="8Gi"),
)

def build_table(n_rows: int, payload_bytes: int) -> pa.Table:
    """An id, a float feature, and a binary payload standing in for an embedding."""
    rng = random.Random(0)
    return pa.table(
        {
            "id": pa.array(range(n_rows), type=pa.int64()),
            "x": pa.array([rng.random() for _ in range(n_rows)], type=pa.float64()),
            "payload": pa.array([rng.randbytes(payload_bytes) for _ in range(n_rows)], type=pa.large_binary()),
        }
    )

and differ only in the return annotation:

@env.task
async def write_parquet(n_rows: int = 100_000, payload_bytes: int = 512) -> DataFrame:
    """No format annotation, so this is stored as Parquet (Flyte's default)."""
    return DataFrame.wrap_df(build_table(n_rows, payload_bytes))

@env.task
async def write_lance(n_rows: int = 100_000, payload_bytes: int = 512) -> Annotated[DataFrame, "lance"]:
    """The same rows, stored as Lance by this plugin."""
    return DataFrame.wrap_df(build_table(n_rows, payload_bytes))

What it compares is how much each side has to materialize to answer the request, via pa.Table.nbytes on the result. This measures decoded size rather than I/O, which is exactly the point: Parquet’s number is much larger because the decoder has to build the entire table first.

@env.task(report=True)
async def compare(parquet_df: DataFrame, lance_ds: lance.LanceDataset, n_random_rows: int = 1_000) -> dict:
    n = lance_ds.count_rows()
    indices = random.Random(7).sample(range(n), k=min(n_random_rows, n))
    columns = ["id", "x", "payload"]

    # `nbytes` is the decoded size of the result, not bytes moved off storage.
    # That is the comparison: how much each side has to build in memory to
    # answer a 1,000-row request.
    #
    # Lance: fetch exactly the requested rows off the open handle.
    lance_bytes = lance_ds.take(indices, columns=columns).nbytes
    lance_seconds = _best_of(lambda: lance_ds.take(indices, columns=columns))

    # Parquet: the decoder is eager, so getting any subset means materializing
    # the whole table first, then indexing into it in memory.
    async def parquet_random():
        table = await parquet_df.open(pa.Table).all()
        table.take(indices)

    parquet_bytes = (await parquet_df.open(pa.Table).all()).nbytes
    parquet_seconds = await _best_of_async(parquet_random)

    await flyte.report.replace.aio(
        _render(n, len(indices), parquet_bytes, lance_bytes, parquet_seconds, lance_seconds),
        do_flush=True,
    )

    return {
        "rows_in_dataset": n,
        "rows_requested": len(indices),
        "parquet_mb_materialized": round(parquet_bytes / 1e6, 1),
        "lance_mb_materialized": round(lance_bytes / 1e6, 2),
        "less_data_x": round(parquet_bytes / lance_bytes, 1),
        "parquet_ms": round(parquet_seconds * 1e3, 1),
        "lance_ms": round(lance_seconds * 1e3, 1),
        "faster_x": round(parquet_seconds / lance_seconds, 1),
    }

Each task publishes the result as a report on the run:

That run is on a cluster, reading from object storage, and it shows the two numbers pulling apart: the materialized-size ratio holds at 100x, while the wall-clock gap is roughly 3.5x. On local disk the same benchmark comes out nearer 10x. Scattered reads against a bucket pay per-request latency that a local file doesn’t, which eats into the advantage without erasing it.

The gap widens as the dataset grows, since Parquet’s cost tracks the dataset while Lance’s tracks the batch, and it narrows as individual rows get fat relative to the batch.

Running the examples

Each example is a self-contained script. Compose the tasks in a driver task and run that:

@env.task
async def main(n: int = 10_000) -> dict:
    ds = await build_dataset(n)

    table_df = await build_table()

    referenced = await convert(n)

    return {
        "summary": await summarize(ds),
        "sampled": await sample_rows(ds),
        "arrow_streaming_rows": await read_as_dataset(table_df),
        "arrow_eager": await read_as_table(table_df),
        "pandas_rows": await as_pandas(table_df),
        "pandas_streaming_rows": await as_pandas_streaming(referenced),
        "referenced": await inspect(referenced),
        "fanned_out_rows": await fan_out(referenced),
    }

if __name__ == "__main__":
    flyte.init_from_config()
    run = flyte.run(main)
    print(run.url)

The other two are shaped the same way. multimodal_streaming.py builds a blob-backed dataset and streams an epoch out of it:

@env.task
async def main(n: int = 4_000) -> dict:
    dataset = await convert(n)
    return {
        "epoch": await train_one_epoch(dataset),
        "labels": await label_histogram(dataset),
        "blob_sizes": await inspect_large_images(dataset),
    }

if __name__ == "__main__":
    flyte.init_from_config()
    run = flyte.run(main)
    print(run.url)

and parquet_vs_lance.py writes both formats and compares them:

@env.task
async def main(n_rows: int = 100_000, payload_bytes: int = 512, n_random_rows: int = 1_000) -> dict:
    parquet_df = await write_parquet(n_rows, payload_bytes)
    lance_df = await write_lance(n_rows, payload_bytes)
    return await compare(parquet_df, lance_df, n_random_rows)

if __name__ == "__main__":
    flyte.init_from_config()
    run = flyte.run(main)
    print(run.url)

Running a script directly with python lance_example.py submits to whatever cluster your Flyte config names. To try the same code against local disk first:

flyte run --local lance_example.py main

Same code path either way. A local run just can’t show you the parts that only exist in a bucket: per-request latency, credential threading, and how big the bytes-read advantage really is on your data.

Common use cases

• Training-data pipelines: convert a swarm of tiny per-sample files into one Lance dataset, then stream shuffled batches from object storage on every epoch, with no per-file connection setup.
• Vector search and RAG: keep embeddings, an ANN index, and the source documents in one dataset that tasks query directly.
• Feature stores and point lookups: fetch individual records by id out of a large table without scanning it.
• Multimodal datasets: image, audio, or video bytes stored beside structured labels, where most stages read only the labels.
• Datasets larger than memory: stream them in bounded batches instead of hand-partitioning them across tasks.