Data Lake Formation：使用 Daft 操作 DLF Lance 表

安裝依賴

python3 -m pip install lance-dlf daft

lance-dlf 會自動安裝 lance_namespace 和 pyarrow 等依賴。

配置 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

匯入 lance_dlf 會自動註冊 dlf namespace：

import lance_namespace
import lance_dlf  # noqa: F401

ns = lance_namespace.connect("dlf", CONFIG)
print(ns.namespace_id())

第一步：擷取表路徑和憑證

通過Daft 讀寫表之前，先通過 describe_table() 從 DLF 擷取表路徑和 OSS 臨時憑證。

from lance_namespace import DescribeTableRequest

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

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

print(desc.location)
print(sorted((desc.storage_options or {}).keys()))

desc 包含兩個關鍵字段：

• desc.location：Lance 表格儲存體路徑，格式為 oss://bucket/path/to/table

• desc.storage_options：OSS 臨時憑證字典

第二步：設定並調用 OSS 憑證環境變數

Daft 底層通過 PyLance 讀寫 OSS。需要在代碼裡定義 apply_oss_environment，把 DLF 返回的臨時憑證設定成 OSS_* 環境變數，然後調用該函數：

import os


def apply_oss_environment(storage_options: dict) -> None:
    os.environ["OSS_ENDPOINT"] = storage_options["oss_endpoint"]
    os.environ["OSS_ACCESS_KEY_ID"] = storage_options["oss_access_key_id"]
    os.environ["OSS_ACCESS_KEY_SECRET"] = storage_options["oss_secret_access_key"]
    if storage_options.get("oss_security_token"):
        os.environ["OSS_SECURITY_TOKEN"] = storage_options["oss_security_token"]
    if storage_options.get("oss_region"):
        os.environ["OSS_REGION"] = storage_options["oss_region"]

# 調用函數
apply_oss_environment(desc.storage_options or {})

第三步：使用Daft讀取表資料

import daft

df = daft.read_lance(desc.location)
df.show()

追加寫入已有表

完成第一步：擷取表路徑和憑證和第二步：設定並調用 OSS 憑證環境變數後，使用 mode="append" 追加資料：

# 前置：串連Catalog、擷取表路徑和憑證、設定並調用OSS憑證環境變數
desc = ns.describe_table(DescribeTableRequest(id=[DATABASE, TABLE]))
apply_oss_environment(desc.storage_options or {})

# 追加寫入
append_df = daft.from_pydict({
    "f0": [204],
    "f1": ["daft-d"],
})

append_df.write_lance(desc.location, mode="append")

# 寫入後讀取驗證：
df2 = daft.read_lance(desc.location)
df2.show()

建立新表並寫入

新表必須先通過 ns.create_table() 建立（該方法同時寫入首批資料），再用 Daft 進行後續讀寫。

# 前置：串連Catalog、擷取表路徑和憑證、設定並調用OSS憑證環境變數
from datetime import datetime
import pyarrow as pa
from lance_namespace import CreateTableRequest, DescribeTableRequest

# 定義Arrow轉換成IPC位元組流
def arrow_table_to_ipc_bytes(table: pa.Table) -> bytes:
    sink = pa.BufferOutputStream()
    with pa.ipc.new_stream(sink, table.schema) as writer:
        writer.write_table(table)
    return sink.getvalue().to_pybytes()


# 建立表並寫入首批資料
table_name = "test_lance_daft_" + datetime.now().strftime("%Y%m%d_%H%M%S")
table_id = [DATABASE, table_name]

rows = {
    "f0": [201, 202, 203],
    "f1": ["daft-a", "daft-b", "daft-c"],
}
arrow_table = pa.table(rows)

create_response = ns.create_table(
    CreateTableRequest(id=table_id),
    arrow_table_to_ipc_bytes(arrow_table),
)
print(create_response.location)

建立完成後，通過 describe_table 擷取憑證，再用 Daft 讀寫：

# 擷取憑證並設定環境變數
desc = ns.describe_table(DescribeTableRequest(id=table_id))
apply_oss_environment(desc.storage_options or {})

# 讀取驗證
df = daft.read_lance(desc.location)
df.show()

# 使用Daft追加資料
append_rows = {
    "f0": [204],
    "f1": ["daft-d"],
}
append_df = daft.from_pydict(append_rows)

meta = append_df.write_lance(desc.location, mode="append")
meta.show()

# 再次讀取確認
appended_df = daft.read_lance(desc.location)
appended_df.show()

預期輸出：

[
    {"f0": 201, "f1": "daft-a"},
    {"f0": 202, "f1": "daft-b"},
    {"f0": 203, "f1": "daft-c"},
    {"f0": 204, "f1": "daft-d"},
]