Vane Data / API 参考

Relation.flat_map

Relation.flat_map 对每个输入行调用一次同步 Python 可调用对象。一行可以产生零行、一行或多行，完整输出结构由 schema 声明。

签名

text

Relation.flat_map(
    function: Callable[..., typing.Any],
    schema: dict[str, sqltypes.DuckDBPyType] | None = None,
    *,
    batch_size: int | None = None,
    output_batch_size: int | None = None,
    min_task_batch_size: int | None = None,
    preserve_compute_batch_boundaries: bool | None = None,
    cpus: float | None = None,
    gpus: float | None = None,
    memory_bytes: int | None = None,
    execution_backend: typing.Literal["subprocess_task", "subprocess_actor", "ray_task", "ray_actor"] | None = None,
    actor_number: int | None = None,
    target_max_batch_bytes: int | None = None,
    task_input_max_bytes: int | None = None,
    output_target_max_bytes: int | None = None,
) -> DuckDBPyRelation

参数

参数	类型	说明	默认值
function	同步函数、绑定方法或可零参数构造的可调用类	接收行字典，返回行字典、同步可迭代对象或 None	必填
schema	非空 dict[str, DuckDBPyType]	完整输出列名、顺序和类型。当前运行时要求显式传入	必填
batch_size	正整数或 None	计算批次的最大输入行数；可调用对象仍逐行接收字典	None
output_batch_size	正整数或 None	输出 Arrow 数据块的目标行数	None
min_task_batch_size	正整数或 None	Task 输入合并的软下限；要求设置 batch_size，且不能小于它	None
preserve_compute_batch_boundaries	bool 或 None	当前接受该参数，但逐行输出路径没有独立的边界刷新语义	None
cpus	非负有限数或 None	每个 Task 或 Actor 的 CPU 资源	None
gpus	非负有限数或 None	每个 Task 或 Actor 的 GPU 资源；正值只支持 Ray	None
memory_bytes	正整数或 None	每个 Task 或 Actor 使用的内存资源；仅 Ray 后端可用	None
execution_backend	subprocess_task、subprocess_actor、ray_task、ray_actor 或 None	执行后端；省略时根据 runner 和可调用对象形态选择	None
actor_number	正整数或 None	Actor 实例数量；Actor 后端必填，Task 后端不能设置	None
target_max_batch_bytes	正整数或 None	Task 输入和输出数据块的共同字节目标	None
task_input_max_bytes	正整数或 None	单次 Task 或 Actor 调用的输入字节目标	None
output_target_max_bytes	正整数或 None	输出数据块的字节目标	None

返回值与错误

flat_map() 返回一个新的 Relation，原 Relation 不会改变。结果只包含 schema 中声明的列。

函数、schema 或执行参数不正确时，调用 flat_map() 就会报错。函数运行失败、返回值格式不正确或某个值不符合 schema 时，会在 fetchall() 等取结果操作中报错。

Task 和 Actor 后端可能重试调用，因此外部副作用必须具备幂等性。可调用类会在相互独立、随时可能重建的 Actor 中运行；任务不保证固定分配给某个 Actor，也没有全局执行顺序，Actor 重建会清空本地状态。

示例

example.py

import vane




def split_words(row):
    for word in row["text"].split():
        yield {"word": word.lower()}




source = vane.sql("SELECT 'Vane Data' AS text")
result = source.flat_map(
    split_words,
    schema={"word": vane.sqltypes.VARCHAR},
)


print(result.fetchall())
vane.close()

输出：

text

[('vane',), ('data',)]

源码与相关页面

编辑本页