E-MapReduce
E-MapReduce
组件与 API 参考
请输入
- 文档首页
E-MapReduce组件操作指南Lance最佳实践EMR Daft 集成 Lance 操作指南实践
复制全文
下载 pdfEMR Daft 集成 Lance 操作指南实践
文档反馈
问问助手简介
Lance 是一种为多模态数据（如图像、视频、音频和通用列式数据）设计的下一代列式存储格式。它支持本地 POSIX 文件系统和云对象存储（如 S3、GCS），以其极快的随机访问、零拷贝读取、与 Daft/DuckDB 的深度集成以及在向量检索工作负载中的卓越性能而著称。
Daft 为 Lance 提供了强大的支持，其核心能力包括：
- 并行与分布式读取：Daft 可以在默认的多线程或分布式的 Ray Runner 上并行执行 Lance 数据集的读取操作。
- 数据跳过（Data Skipping）：利用 df.where() 谓词和文件/片段的统计信息，Daft 能够高效跳过不匹配的数据，加速查询。
- 版本与时间切片读取：通过 version 和 asof 参数，你可以精确读取 Lance 数据集的特定版本或某个时间点之前的最新版本。
- 扫描与缓存优化：支持通过 fragment_group_size 调整扫描效率，并通过 index_cache_size、metadata_cache_size_bytes 等参数微调缓存行为，优化大规模数据集的性能。
本指南将详细介绍如何使用 Daft 读写 Lance 数据集，并重点讲解如何结合火山引擎 TOS (S3 兼容) 进行配置，提供在仅安装 pylance 和 daft 的环境下可直接复制运行的最佳实践代码。
安装与环境准备
- 开通 Serverless EMR：Serverless EMR 当前支持 Daft/Ray 计算引擎，可以使用 serverless Ray 提交作业，作业提交方式参考 Serverless Ray 使用指南分布式的daft运行在 ray引擎上。
- 开通TOS服务用于数据存储。
EMR云上环境
- EMR的Serverless Ray镜像默认已安装daft和pylance，用户无需单独安装。
- 如果需要在自己的机器上安装单机版本并验证代码正确性，操作如下。
- 如果你的镜像没有自带daft，可以参考构建和使用自定义镜像 进行自定义镜像安装对应的包。
构建本地开发环境
可以按照以下步骤准备您的本地 Python 环境，用于单机调试代码
# 安装核心依赖：Daft 和 Lance Python SDK (pylance)
pip install -U pylance daft
TOS 环境配置
S3/TOS 配置与访问
Daft 提供了灵活的 S3 配置方式，支持标准凭证链和手动的 IOConfig 配置。
凭证发现机制
- 环境/CLI 凭证链（推荐）：Daft 会自动查找标准 AWS 环境变量（AWS_ACCESS_KEY_ID 等）或 AWS CLI 配置文件（~/.aws/credentials）。
- 手动 IOConfig：构建一个 IOConfig 对象，在 read_lance 或 write_lance 调用中通过 io_config 参数传入。
以下代码片段定义了访问火山引擎 TOS 所需的配置。建议通过环境变量注入敏感信息，此处为方便演示，提供了测试用的默认值。后续所有示例将复用此配置。
import os
import daft
from daft.io import IOConfig, S3Config
# 建议使用环境变量注入，以下为测试默认值（仅用于PoC演示）：
AWS_ENDPOINT = os.getenv("AWS_ENDPOINT", "https://tos-s3-cn-shanghai.volces.com") # 无bucket前缀
AWS_KEY_ID = os.getenv("AWS_KEY_ID", "your_ak")
AWS_ACCESS_KEY = os.getenv("AWS_ACCESS_KEY", "your_sk")
AWS_REGION = os.getenv("AWS_REGION", "cn-beijing")
BUCKET = os.getenv("BUCKET", "your bucket name")
# Daft 读写时使用的 io_config
io_config = IOConfig(
s3=S3Config(
endpoint_url=AWS_ENDPOINT,
region_name=AWS_REGION,
key_id=AWS_KEY_ID,
access_key=AWS_ACCESS_KEY,
force_virtual_addressing=True, # 自动重写为 bucket.endpoint 形式
verify_ssl=True,
)
)
# 统一的 TOS 路径前缀
TOS = f"s3://{BUCKET}"
S3Config 关键参数详解
S3Config 提供了丰富的参数来定制 S3 访问行为：
- region_name (str): AWS 区域，例如 "us-west-2"。
- endpoint_url (str): S3 端点 URL。对于 AWS S3，通常不需要设置。对于 S3 兼容存储（如 TOS），此为必填项。
- key_id (str): Access Key ID。
- access_key (str): Secret Access Key。
- force_virtual_addressing (bool): 是否强制使用虚拟主机风格的地址（bucket.endpoint）。某些 S3 兼容服务需要设为 True。当它为 True 时，Daft 内部会将 endpoint_url 从 https://host 转换为 https://bucket.host 的形式。
读取 Lance 数据集
使用 daft.read_lance() 函数可以轻松读取云端的 Lance 数据集。
基础读取
此示例首先使用 daft.from_pydict 创建一个 Daft DataFrame，然后用 df.write_lance 将其写入 TOS，最后使用 daft.read_lance 将其读回。
import daft
# (复用上一节的配置)
# 1. 创建一个 Daft DataFrame
df = daft.from_pydict({"a": [1, 2, 3], "b": ["x", "y", "z"]})
# 2. 将数据写入 TOS (覆盖写)
path = f"{TOS}/demo/my_lance_dataset"
df.write_lance(path, io_config=io_config, mode="overwrite")
# 3. 从 TOS 读取
print("### 从 TOS 读取数据集:")
df_tos = daft.read_lance(path, io_config=io_config)
df_tos.show()
运行结果
以下为实际运行的输出片段，仅供参考。
### 从 TOS 读取数据集:
╭───────┬────────╮
│ a ┆ b │
│ --- ┆ --- │
│ Int64 ┆ String │
╞═══════╪════════╡
│ 1 ┆ x │
├╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌┤
│ 2 ┆ y │
├╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌┤
│ 3 ┆ z │
╰───────┴────────╯
(Showing first 3 of 3 rows)
指定版本与时间切片
Lance 支持数据版本控制。你可以使用 version 参数读取特定版本，或使用 asof 参数进行时间点查询。
import daft
from datetime import datetime
# (复用上一节的配置)
# 1. 创建并写入两个版本的数据集
base_path = f"{TOS}/demo/versioned_dataset"
df_v1 = daft.from_pydict({"a": [1, 2]})
df_v1.write_lance(base_path, mode="overwrite", io_config=io_config)
df_v2 = daft.from_pydict({"a": [3, 4]})
df_v2.write_lance(base_path, mode="append", io_config=io_config)
# 2. 读取最新版本 (版本 2)
print("### 读取最新版本:")
df_latest = daft.read_lance(base_path, io_config=io_config)
df_latest.show()
# 3. 按版本号读取 (版本 1)
print("\n### 按版本号读取 (版本 1):")
df_v1_read = daft.read_lance(base_path, version=1, io_config=io_config)
df_v1_read.show()
# 4. 按时间点读取 (asof) - 示例
# asof 参数需要一个 datetime 对象或符合 RFC3339/ISO8601 的字符串
# 假设在 v2 写入前的时间点
# df_asof = daft.read_lance(base_path, asof="2025-12-01T00:00:00Z", io_config=io_config)
# print("\n### 按时间点读取 (asof):")
# df_asof.show() # 预期只显示版本 1 的数据
运行结果
以下为实际运行的输出片段，仅供参考。
### 读取最新版本:
╭───────╮
│ a │
│ --- │
│ Int64 │
╞═══════╡
│ 1 │
├╌╌╌╌╌╌╌┤
│ 2 │
├╌╌╌╌╌╌╌┤
│ 3 │
├╌╌╌╌╌╌╌┤
│ 4 │
╰───────╯
(Showing first 4 of 4 rows)
### 按版本号读取 (版本 1):
╭───────╮
│ a │
│ --- │
│ Int64 │
╞═══════╡
│ 1 │
├╌╌╌╌╌╌╌┤
│ 2 │
╰───────╯
(Showing first 2 of 2 rows)
过滤、投影与 Limit 下推
Daft 会自动将 where()、select() 和 limit() 操作下推到 Lance 扫描器，以实现高效的数据过滤、列选择和行数限制。你可以通过 explain(True) 查看物理计划，确认下推是否生效。
import daft
from daft.expressions import col
# (复用上一节的配置)
# 1. 准备数据
base_path = f"{TOS}/demo/pushdown_dataset"
df = daft.from_pydict({"id": list(range(100)), "value": [i * 0.1 for i in range(100)]})
df.write_lance(base_path, mode="overwrite", io_config=io_config)
# 2. 开启严格过滤下推
daft.context.set_planning_config(enable_strict_filter_pushdown=True)
# 3. 读取并应用过滤、投影和 Limit
df_read = daft.read_lance(base_path, io_config=io_config)
filtered_df = df_read.where(col("id") > 50).select("value").limit(5)
# 4. 查看执行计划
print("### 执行计划:")
filtered_df.explain(True)
# 预期在 explain 的输出中看到 Pushdowns 信息，表明操作已下推
# LanceScanOperator:
# Pushdowns:
# filter: (col(id) > 50)
# limit: 5
# projection: [value]
运行结果
以下为实际运行的输出片段，仅供参考。
### 执行计划:
== Unoptimized Logical Plan ==
* Limit: 5
|
* Project: col(value)
|
* Filter: col(id) > lit(50)
|
* PythonScanOperator: LanceDBScanOperator(s3://lance-poc/demo/pushdown_dataset)
| File schema = id#Int64, value#Float64
| Partitioning keys = []
| Output schema = id#Int64, value#Float64
== Optimized Logical Plan ==
* Num Scan Tasks = 1
| File schema = id#Int64, value#Float64
| Partitioning keys = []
| Projection pushdown = [value]
| Filter pushdown = col(id) > lit(50)
| Limit pushdown = 5
| Output schema = value#Float64
| Stats = { Approx num rows = 0, Approx size bytes = 0 B, Accumulated selectivity
| = 0.30 }
== Physical Plan ==
* ScanTaskSource:
| Num Scan Tasks = 1
| Estimated Scan Bytes = 0
| Num Parallel Scan Tasks = 1
| Pushdowns: {projection: [value], filter: col(id) > lit(50), limit: 5}
| Schema: {id#Int64, value#Float64}
| Scan Tasks: [
| {daft.io.lance.lance_scan:_lancedb_table_factory_function}
| ]
| Stats = { Approx num rows = 0, Approx size bytes = 0 B, Accumulated selectivity
| = 0.30 }
| Batch Size = Range(0, 131072]
优化扫描：fragment_group_size
对于包含大量小文件（fragments）的 Lance 数据集，频繁的 I/O 和任务调度会带来显著开销。通过设置 fragment_group_size 参数，你可以将多个 fragments 分组到单个扫描任务中处理，从而有效减少开销。
import daft
# (复用上一节的配置)
# 1. 创建一个包含多个 fragment 的数据集
base_path = f"{TOS}/demo/fragments_dataset"
for i in range(10):
df = daft.from_pydict({"id": [i]})
df.write_lance(base_path, mode="append" if i > 0 else "overwrite", io_config=io_config)
# 2. 不分组读取，每个 fragment 对应一个分区
df_no_grouping = daft.read_lance(base_path, io_config=io_config)
print(f"不分组的分区数: {df_no_grouping.num_partitions()}") # 预期为 10
# 3. 将 3 个 fragment 分为一组
df_grouped = daft.read_lance(base_path, fragment_group_size=3, io_config=io_config)
print(f"分组后的分区数: {df_grouped.num_partitions()}") # 预期为 4 (ceil(10/3))
运行结果
以下为实际运行的输出片段，仅供参考。
不分组的分区数: 10
分组后的分区数: 4
写入 Lance 数据集
使用 DataFrame.write_lance() 方法可以将 Daft DataFrame 写入 Lance 格式。
写入模式与文件切分参数
write_lance 支持三种写入模式，并提供参数控制文件切分，这对于治理小文件至关重要。
- mode:
- create: 创建新数据集，若路径已存在则失败。
- append: 向现有数据集追加数据，若路径不存在则失败。
- overwrite: 覆盖或创建新数据集。
- 文件切分参数:
- max_rows_per_file (int): 每个 Lance 文件的最大行数。
- max_bytes_per_file (int): 每个 Lance 文件的最大字节数（近似值）。
- max_rows_per_group (int): 每个 Row Group 的最大行数。Row Group 是 Lance 内部的读写单元。
import daft
# (复用上一节的配置)
path = f"{TOS}/demo/my_lance_table.lance"
df = daft.from_pydict({"a": list(range(10000))})
# 1. 创建新表，并限制每个文件最多 8192 行
print("### 创建模式 (max_rows_per_file):")
meta_create = df.write_lance(path, mode="create", io_config=io_config, max_rows_per_file=8192)
meta_create.show()
# 2. 覆盖写，并限制每个文件最大 1KB (仅为演示，实际应设更大值)
print("\n### 覆盖模式 (max_bytes_per_file):")
meta_overwrite = df.write_lance(path, mode="overwrite", io_config=io_config, max_bytes_per_file=1024)
meta_overwrite.show()
# 3. 追加写，并限制每个 Row Group 最多 1024 行
print("\n### 追加模式 (max_rows_per_group):")
meta_append = df.write_lance(path, mode="append", io_config=io_config, max_rows_per_group=1024)
meta_append.show()
运行结果
以下为实际运行的输出片段，仅供参考。
### 创建模式 (max_rows_per_file):
╭───────────────┬──────────────────┬─────────────────┬─────────╮
│ num_fragments ┆ num_deleted_rows ┆ num_small_files ┆ version │
│ --- ┆ --- ┆ --- ┆ --- │
│ Int64 ┆ Int64 ┆ Int64 ┆ Int64 │
╞═══════════════╪══════════════════╪═════════════════╪═════════╡
│ 2 ┆ 0 ┆ 0 ┆ 1 │
╰───────────────┴──────────────────┴─────────────────┴─────────╯
(Showing first 1 of 1 rows)
### 覆盖模式 (max_bytes_per_file):
╭───────────────┬──────────────────┬─────────────────┬─────────╮
│ num_fragments ┆ num_deleted_rows ┆ num_small_files ┆ version │
│ --- ┆ --- ┆ --- ┆ --- │
│ Int64 ┆ Int64 ┆ Int64 ┆ Int64 │
╞═══════════════╪══════════════════╪═════════════════╪═════════╡
│ 1 ┆ 0 ┆ 0 ┆ 2 │
╰───────────────┴──────────────────┴─────────────────┴─────────╯
(Showing first 1 of 1 rows)
### 追加模式 (max_rows_per_group):
╭───────────────┬──────────────────┬─────────────────┬─────────╮
│ num_fragments ┆ num_deleted_rows ┆ num_small_files ┆ version │
│ --- ┆ --- ┆ --- ┆ --- │
│ Int64 ┆ Int64 ┆ Int64 ┆ Int64 │
╞═══════════════╪══════════════════╪═════════════════╪═════════╡
│ 2 ┆ 0 ┆ 0 ┆ 3 │
╰───────────────┴──────────────────┴─────────────────┴─────────╯
(Showing first 1 of 1 rows)
Schema 控制
write_lance 允许你通过 Daft 的 Schema 对象来精细控制写入的 schema，包括类型、顺序和可空性。
from daft_tos_config import io_config, TOS
import daft
df = daft.from_pydict({
"id": [1],
"compressible_text": ["a" * 200],
"image_bytes": [b"very-large-image-data..."]
})
path = f"{TOS}/demo/schema_lance_dataset"
df.write_lance(path, mode="overwrite", io_config=io_config)
df_read = daft.read_lance(path, io_config=io_config)
print("### 写入后的 Schema:")
print(df_read.schema())
运行结果
以下为实际运行的输出片段，仅供参考。
### 写入后的 Schema:
╭───────────────────┬────────╮
│ column_name ┆ type │
╞═══════════════════╪════════╡
│ id ┆ Int64 │
├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌┤
│ compressible_text ┆ String │
├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌┤
│ image_bytes ┆ Binary │
╰───────────────────┴────────╯
说明
关于高级编码元数据高级编码元数据（例如 zstd 压缩、Blob 存储等）的指定需要通过 PyArrow 的 Schema 元数据能力。若您的环境仅安装了 pylance 和 daft，本示例将不展示该功能，以确保代码的普适性。如果需要此功能，请安装 pyarrow 并参考相关文档。
补充说明：高级编码元数据（如 zstd 压缩、Blob）需要 PyArrow 的元数据能力；在仅安装 pylance+daft 的本地环境中，示例展示默认推断的 schema。
Lance 文件切分与小文件治理
核心原则：避免产生过多的小文件（碎片）。大量小文件会导致 I/O 开销剧增、元数据管理复杂化，并最终导致查询性能严重下降，尤其是在对象存储上。
最佳实践推荐
维度 | 推荐区间 | 说明 |
行数/文件 | 8,000 – 1,000,000 | 根据列宽和查询模式调整。对于宽表或扫描密集型任务，倾向于更大的文件。一般不建议低于 4096 行。 |
大小/文件 | 64MB – 512MB | 这是对象存储的最佳实践范围。过小的文件会显著增加 API 请求开销和元数据读取延迟。 |
写入参数建议
- 使用 Daft (df.write_lance)：
- max_bytes_per_file: 首选参数。直接控制文件大小，最符合对象存储的最佳实践。
- max_rows_per_file: 当数据行大小比较均匀时，可作为备选。
- max_rows_per_group: 控制 Row Group 粒度。一个文件可以包含多个 Row Group。合理的 Row Group 大小（通常建议 20,000 - 100,000 行）有助于过滤下推。
读取与治理策略
- 读取优化：如果已存在大量小 fragment，在读取侧使用 daft.read_lance 的 fragment_group_size 参数。它可以将多个小文件合并成一个扫描任务，减少 I/O 和调度开销。
- 离线治理：定期对碎片化的数据集进行重写（Compaction）。在重写时，统一规划写入参数（如 max_bytes_per_file）和压缩策略，生成结构健康的新版本。
常见问答与排查
- Q: 遇到 403 Forbidden 或 404 Not Found 错误？
- A: 通常是权限问题。请检查凭证、存储桶策略和 IAM 策略。
- Q: create 模式写入失败？
- A: 确保目标路径不是一个已存在的文件或 Lance 数据集。
- Q: append 模式写入失败？
- A: 确保目标路径是一个已存在的 Lance 数据集，并且待追加数据的 schema 与目标表兼容。
最近更新时间：2026.01.19 17:09:49
这个页面对您有帮助吗？
有用
有用
无用
无用