Data Lake Formation:Daft を使用した DLF Lance テーブルの操作

依存関係のインストール

python3 -m pip install lance-dlf daft        

lance-dlf は、lance_namespace、pyarrow、およびその他の必要な依存関係を自動的にインストールします。

カタログ接続の設定

CONFIG = {
    "uri": "http://<dlf-endpoint>", # パブリックアクセスの場合は HTTPS プロトコルを使用
    "warehouse": "<warehouse>",
    "token.provider": "dlf",
    "dlf.region": "<region>",
    "dlf.access-key-id": "<access-key-id>",
    "dlf.access-key-secret": "<access-key-secret>",
    "dlf.oss-endpoint": "<oss-endpoint>",
}

DLF への接続

lance_dlf をインポートすると、dlf 名前空間が自動的に登録されます：

import lance_namespace
import lance_dlf  # noqa: F401

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

ステップ 1：テーブルパスと認証情報の取得

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 には、次の 2 つの主要なフィールドが含まれています：

• desc.location — oss://bucket/path/to/table 形式の Lance テーブルのストレージパス

• desc.storage_options — 一時的な OSS 認証情報のディクショナリ

ステップ 2：OSS 認証情報環境変数の設定

Daft は内部で PyLance を使用して OSS にアクセスします。DLF が発行した認証情報を OSS_* 環境変数にマッピングするため、ヘルパー関数 apply_oss_environment を定義して呼び出します。

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 {})        

ステップ 3：Daft でのテーブルデータの読み取り

import daft

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

既存テーブルへの追加

上記の「ステップ 1：テーブルパスと認証情報の取得」と「ステップ 2：OSS 認証情報環境変数の設定」を完了した後、mode="append" を使用してデータを追加します：

# 前提条件：カタログへの接続、テーブルパス/認証情報の取得、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 を使用します。

# 前提条件：カタログへの接続、テーブルパス/認証情報の取得、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"}
]