在图片、音视频和压缩包等数据处理场景中，单个对象可能占用较多内存。本文介绍如何使用 Lance-Ray 按批次处理并写入 Lance，避免在应用中先收集全部处理结果。

write_lance 是 Ray Dataset 写入 Lance 的高层接口。例如，Worker 可以逐个读取视频并提取特征，释放不再使用的视频缓冲区，只把较小的特征和元数据交给后续写入流程。

工作原理 #

1. Worker 读取源对象并完成处理，将较小的结果交给 Ray Dataset。
2. 分布式写入任务将结果写入 Lance 数据文件。
3. 所有写入完成后，Driver 汇总 Fragment 元数据并提交新版本。

示例代码 #

以下示例使用只包含 source_uri 字符串列的 Ray Dataset 作为输入，source_uris 和 sources 也用于后续示例。请将示例中的读取和处理逻辑替换为实际业务逻辑。TARGET_URI 和 STORAGE_OPTIONS 分别表示目标数据集路径和存储配置。

import ray

# 替换为实际的源对象 URI。
source_uris = ["tos://source-bucket/object-1", "tos://source-bucket/object-2"]
sources = ray.data.from_items([{"source_uri": uri} for uri in source_uris])

本例每批处理 32 个源对象引用。

import lance_ray as lr
import pyarrow as pa


OUTPUT_SCHEMA = pa.schema(
    {
        "source_uri": pa.string(),
        "object_size": pa.int64(),
        "summary": pa.string(),
    }
)


def process_large_objects(batch: pa.Table) -> pa.Table:
    rows = []
    for source_uri in batch["source_uri"].to_pylist():
        # large_object = read_large_object(source_uri)
        # object_size, summary = process(large_object)
        object_size, summary = 0, "result"
        rows.append(
            {
                "source_uri": source_uri,
                "object_size": object_size,
                "summary": summary,
            }
        )
    return pa.Table.from_pylist(rows, schema=OUTPUT_SCHEMA)


results = sources.map_batches(
    process_large_objects,
    batch_format="pyarrow",
    batch_size=32,
    concurrency=4,
)

lr.write_lance(
    results,
    uri=TARGET_URI,
    schema=OUTPUT_SCHEMA,
    mode="overwrite",
    min_rows_per_file=1024,
    storage_options=STORAGE_OPTIONS,
    concurrency=4,
)

参数说明 #

当处理结果本身仍然很大时，将其作为中间 Ray Dataset 在任务间传递，会增加 Ray Object Store 的序列化、内存和 spill 压力。LanceFragmentWriter 可以把转换与写入放在同一个 Worker 内，避免将大结果作为 map_batches 的输出返回给 Ray。

工作原理 #

1. Worker 接收源对象引用，处理函数通过生成器逐批产出结果。
2. LanceFragmentWriter 在同一个 Worker 内直接写入这些结果，向 Ray 返回 Fragment 元数据。
3. LanceFragmentCommitter 汇总写入结果，由 Driver 提交新版本。

示例代码 #

本例一次最多接收 32 个 URI，在生成器中逐个处理，每次产出一行结果。

import lance
import pyarrow as pa
from lance_ray.datasink import LanceFragmentCommitter
from lance_ray.fragment import LanceFragmentWriter


OUTPUT_SCHEMA = pa.schema(
    {
        "source_uri": pa.string(),
        "payload": pa.large_binary(),
    }
)


def process_large_objects(batch: pa.Table):
    for source_uri in batch["source_uri"].to_pylist():
        # large_object = read_large_object(source_uri)
        # payload = process(large_object)
        payload = b"result" # 大对象占位值
        yield pa.Table.from_pylist(
            [{"source_uri": source_uri, "payload": payload}],
            schema=OUTPUT_SCHEMA,
        )


# 先创建空表，再追加处理结果。
lance.write_dataset(
    pa.Table.from_pylist([], schema=OUTPUT_SCHEMA),
    TARGET_URI,
    mode="create",
    storage_options=STORAGE_OPTIONS,
)

fragment_metadata = sources.map_batches(
    LanceFragmentWriter(
        TARGET_URI,
        schema=OUTPUT_SCHEMA,
        transform=process_large_objects,
        storage_options=STORAGE_OPTIONS,
    ),
    batch_format="pyarrow",
    batch_size=32,
    concurrency=2,
)

fragment_metadata.write_datasink(
    LanceFragmentCommitter(
        TARGET_URI,
        mode="append",
        storage_options=STORAGE_OPTIONS,
    ),
)

单机处理时，可以将生成器产出的批次通过 RecordBatchReader 写入 Lance。

import lance
import pyarrow as pa


OUTPUT_SCHEMA = pa.schema(
    {
        "source_uri": pa.string(),
        "payload": pa.large_binary(),
    }
)


def batches():
    for source_uri in source_uris:
        # large_object = read_large_object(source_uri)
        # payload = process(large_object)
        payload = b"result"
        yield pa.RecordBatch.from_pylist(
            [{"source_uri": source_uri, "payload": payload}],
            schema=OUTPUT_SCHEMA,
        )


lance.write_dataset(
    pa.RecordBatchReader.from_batches(OUTPUT_SCHEMA, batches()),
    TARGET_URI,
    mode="overwrite",
    storage_options=STORAGE_OPTIONS,
)

• EMR Lance-Ray 通用操作指导：对象存储连接与基础读写。
• Ray Data map_batches 文档：输入 batch 与调度语义。