使用 Python 操作 DLF Lance 表

安装 lance-dlf

python3 -m pip install lance-dlf

配置 Catalog 连接

CONFIG = {
    "uri": "http://<DLF-ENDPOINT>", # 公网访问须使用HTTPS协议
    "warehouse": "<YOUR-CATALOG>",
    "token.provider": "dlf",
    "dlf.region": "<REGION-ID>",
    "dlf.access-key-id": "<ACCESS-KEY-ID>",
    "dlf.access-key-secret": "<ACCESS-KEY-SECRET>",
    "dlf.oss-endpoint": "<OSS-ENDPOINT>", # VPC访问可选，公网访问必填
}

连接 DLF Catalog

导入 lance_dlf 模块会自动注册 dlf namespace 实现：

import lance_namespace
import lance_dlf  # noqa: F401

# 连接 DLF Catalog
ns = lance_namespace.connect("dlf", CONFIG)

# 验证连接
print(ns.namespace_id())

namespace_id() 方法返回当前连接的 Catalog 信息，用于确认连接的 DLF 访问端点和 Catalog名称。

查看 Namespace 和 Table

from lance_namespace import (
    DescribeNamespaceRequest,
    DescribeTableRequest,
    ListNamespacesRequest,
    ListTablesRequest,
)

DATABASE = "<database>"
TABLE = "<table>"

# 列出所有 Namespace
namespaces = ns.list_namespaces(ListNamespacesRequest(id=[]))
print(namespaces)

# 查看指定 Namespace 元数据
namespace = ns.describe_namespace(DescribeNamespaceRequest(id=[DATABASE]))
print(namespace)

# 列出 Namespace 下的所有 Table
tables = ns.list_tables(ListTablesRequest(id=[DATABASE]))
print(tables)

# 查看 Table 详情
table = ns.describe_table(DescribeTableRequest(id=[DATABASE, TABLE]))
print(table.location)
print(table.properties)
print(table.storage_options)

创建 Lance 表并写入数据

如果表不存在，需要先创建表然后写入数据。原理：

1. 根据 Arrow Schema 构造 DLF Schema，并使用低层 API 在 DLF 中创建表

2. 创建完成后获取 Lance 表的存储位置（location）

3. 构造 OSS 临时访问凭证（storage_options）

4. 使用 PyLance 将 Arrow 数据写入指定的存储位置，然后通知 DLF 提交 Lance 写入。

import lance
import pyarrow as pa
from lance_dlf.api.rest_exception import AlreadyExistsException
from lance_dlf.common.identifier import Identifier
from lance_dlf.schema.schema import Schema as DlfSchema

DATABASE = "default"
TABLE = "test_lance_create_001"

table_id = [DATABASE, TABLE]

# 构造测试数据
data = pa.table({
    "f0": pa.array([101, 102, 103], type=pa.int64()),
    "f1": pa.array(["create-a", "create-b", "create-c"], type=pa.string()),
})

# 创建表并写入数据
expected = DlfSchema.from_pyarrow_schema(data.schema, options={"type": "lance-table"})
identifier = Identifier(*table_id)

try:
    ns._api.create_table(identifier, expected)
except AlreadyExistsException:
    response = ns._api.get_table(identifier)
    if response.get_schema() != expected:
        raise ValueError("已存在表的 Schema 与待写入数据的 Schema 不一致")
else:
    response = ns._api.get_table(identifier)

location = response.get_path()
storage_options = ns._build_storage_options(identifier)
lance.write_dataset(data, location, mode="overwrite", storage_options=storage_options)
ns.notify_lance_commit(table_id)

# 读取验证
dataset = lance.dataset(location, storage_options=storage_options)
result = dataset.to_table()
print(result)

pyarrow.Table
f0: int64
f1: string
----
f0: [[101,102,103]]
f1: [["create-a","create-b","create-c"]]

写入已有表

如果 DLF 中已存在 Lance 表，可以先通过 describe_table 获取表的存储位置和访问凭证，然后使用 PyLance 写入数据。

import lance
import pyarrow as pa
from lance_namespace import DescribeTableRequest

DATABASE = "default"
TABLE = "test_lance_table"

table_id = [DATABASE, TABLE]

# 获取表详情
desc = ns.describe_table(DescribeTableRequest(id=table_id))

# 构造测试数据
data = pa.table({
    "f0": pa.array([1, 2, 3], type=pa.int64()),
    "f1": pa.array(["value-1", "value-2", "value-3"], type=pa.string()),
})

# 写入数据
lance.write_dataset(
    data,
    desc.location,
    mode="overwrite",
    storage_options=desc.storage_options,
)
ns.notify_lance_commit(table_id)

# 读取验证
dataset = lance.dataset(desc.location, storage_options=desc.storage_options)
print(dataset.to_table())

读取数据

如果 DLF 中已存在 type=lance-table 的表且已包含数据，可以通过 describe_table 获取存储位置和临时凭证，然后使用 PyLance 读取数据。详见创建 Lance 表并写入数据代码块末端的“读取验证”部分。

自动构造测试数据

如果不想硬编码列名和类型，可以从 DLF Table Schema 中提取字段信息，自动构造测试数据。

import pyarrow as pa
from lance_dlf.common.identifier import Identifier


def sample_value(field_type: str, row: int):
    """根据字段类型生成示例值"""
    normalized = field_type.lower()
    if "int" in normalized:
        return row + 1
    if "string" in normalized or "char" in normalized or "varchar" in normalized:
        return f"value-{row + 1}"
    raise ValueError(f"Unsupported sample field type: {field_type}")


def build_sample_table(ns, database: str, table: str) -> pa.Table:
    """根据 DLF 表结构构造示例数据"""
    raw_table = ns._api.get_table(Identifier(database, table))
    schema = raw_table.get_schema()
    fields = schema.fields if schema and schema.fields else []

    data = {}
    for field in fields:
        field_type = str(field.type)
        data[field.name] = [sample_value(field_type, row) for row in range(3)]

    return pa.table(data)

desc = ns.describe_table(DescribeTableRequest(id=[DATABASE, TABLE]))
data = build_sample_table(ns, DATABASE, TABLE)

lance.write_dataset(
    data,
    desc.location,
    mode="overwrite",
    storage_options=desc.storage_options,
)