## 简介 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 使用指南](https://www.volcengine.com/docs/6491/1353774?lang=zh)分布式的daft运行在 ray引擎上。 * [开通TOS服务](https://www.volcengine.com/docs/6349/74830?lang=zh)用于数据存储。 ### EMR云上环境 * EMR的Serverless Ray镜像默认已安装daft和pylance，用户无需单独安装。 * 如果需要在自己的机器上安装单机版本并验证代码正确性，操作如下。 * 如果你的镜像没有自带daft，可以参考[构建和使用自定义镜像](https://www.volcengine.com/docs/6491/1348513?lang=zh) 进行自定义镜像安装对应的包。 ### 构建本地开发环境可以按照以下步骤准备您的本地 Python 环境，用于单机调试代码 ```Bash # 安装核心依赖：Daft 和 Lance Python SDK (pylance) pip install -U pylance daft ``` ## TOS 环境配置 ### S3/TOS 配置与访问 Daft 提供了灵活的 S3 配置方式，支持标准凭证链和手动的 `IOConfig` 配置。 ### 凭证发现机制 1. **环境/CLI 凭证链（推荐）** ：Daft 会自动查找标准 AWS 环境变量（`AWS_ACCESS_KEY_ID` 等）或 AWS CLI 配置文件（`~/.aws/credentials`）。 2. **手动** **`IOConfig`**：构建一个 `IOConfig` 对象，在 `read_lance` 或 `write_lance` 调用中通过 `io_config` 参数传入。以下代码片段定义了访问火山引擎 TOS 所需的配置。建议通过环境变量注入敏感信息，此处为方便演示，提供了测试用的默认值。后续所有示例将复用此配置。 ```Python 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` 将其读回。 ```Python 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() ``` #### 运行结果以下为实际运行的输出片段，仅供参考。 ```Plain ### 从 TOS 读取数据集: ╭───────┬────────╮ │ a ┆ b │ │ --- ┆ --- │ │ Int64 ┆ String │ ╞═══════╪════════╡ │ 1 ┆ x │ ├╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌┤ │ 2 ┆ y │ ├╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌┤ │ 3 ┆ z │ ╰───────┴────────╯ (Showing first 3 of 3 rows) ``` ### 指定版本与时间切片 Lance 支持数据版本控制。你可以使用 `version` 参数读取特定版本，或使用 `asof` 参数进行时间点查询。 ```Python 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 的数据 ``` #### 运行结果以下为实际运行的输出片段，仅供参考。 ```Plain ### 读取最新版本: ╭───────╮ │ 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)` 查看物理计划，确认下推是否生效。 ```Python 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] ``` #### 运行结果以下为实际运行的输出片段，仅供参考。 ```Plain ### 执行计划: == 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 分组到单个扫描任务中处理，从而有效减少开销。 ```Python 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)) ``` #### 运行结果以下为实际运行的输出片段，仅供参考。 ```Plain 不分组的分区数: 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 内部的读写单元。 ```Python 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() ``` #### 运行结果以下为实际运行的输出片段，仅供参考。 ```Plain ### 创建模式 (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，包括类型、顺序和可空性。 ```Python 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()) ``` #### 运行结果以下为实际运行的输出片段，仅供参考。 ```Plain ### 写入后的 Schema: ╭───────────────────┬────────╮ │ column_name ┆ type │ ╞═══════════════════╪════════╡ │ id ┆ Int64 │ ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌┤ │ compressible_text ┆ String │ ├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌┤ │ image_bytes ┆ Binary │ ╰───────────────────┴────────╯ ``` :::tip **关于高级编码元数据**高级编码元数据（例如 `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 与目标表兼容。

E-MapReduce