Data Lake Formation:Gunakan tabel Lance DLF dengan Python

Instal lance-dlf

Instal lance-dlf dari PyPI:

python3 -m pip install lance-dlf

Konfigurasi koneksi katalog

CONFIG = {
    "uri": "http://<dlf-endpoint>", # Untuk akses publik, gunakan protokol 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>",
}

Hubungkan ke katalog DLF

Mengimpor modul lance_dlf secara otomatis mendaftarkan implementasi namespace dlf:

import lance_namespace
import lance_dlf  # noqa: F401

# Hubungkan ke DLF Catalog
ns = lance_namespace.connect("dlf", CONFIG)

# Verifikasi koneksi
print(ns.namespace_id())

Metode namespace_id() mengembalikan informasi tentang katalog yang terhubung, termasuk endpoint DLF dan warehouse.

Buat tabel Lance baru dan tulis data

Jika tabel belum ada, buat tabel dan masukkan data. Mekanisme:

1. Buat tabel menggunakan ns.create_table()

2. Dapatkan lokasi penyimpanan tabel Lance (location) dari DLF

3. Dapatkan kredensial akses OSS sementara melalui lance-dlf

4. Konversi data Arrow ke byte IPC dan tulis data ke lokasi yang ditentukan dengan PyLance (lihat Referensi)

import lance
import pyarrow as pa
from lance_namespace import CreateTableRequest, DescribeTableRequest

DATABASE = "default"
TABLE = "test_lance_create_001"

table_id = [DATABASE, TABLE]

# Bangun data uji
data = pa.table({
    "f0": pa.array([101, 102, 103], type=pa.int64()),
    "f1": pa.array(["create-a", "create-b", "create-c"], type=pa.string()),
})

# Konversi tabel Arrow ke byte IPC
def arrow_table_to_ipc_bytes(table):
    sink = pa.BufferOutputStream()
    writer = pa.ipc.new_stream(sink, table.schema)
    writer.write_table(table)
    writer.close()
    return sink.getvalue().to_pybytes()

# Buat tabel dan tulis data
create_response = ns.create_table(
    CreateTableRequest(id=table_id),
    arrow_table_to_ipc_bytes(data),
)

print(create_response.location)
print(create_response.storage_options.keys())

# Baca dan verifikasi
desc = ns.describe_table(DescribeTableRequest(id=table_id))
dataset = lance.dataset(desc.location, storage_options=desc.storage_options)
result = dataset.to_table()
print(result)

Output yang Diharapkan

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

Tulis ke tabel kosong yang sudah ada

Jika tabel type=lance-table kosong sudah ada di DLF, dapatkan lokasi penyimpanan dan kredensial akses tabel tersebut melalui describe_table, lalu tulis data dengan PyLance.

import lance
import pyarrow as pa
from lance_namespace import DescribeTableRequest

DATABASE = "default"
TABLE = "test_lance_table"

table_id = [DATABASE, TABLE]

# Dapatkan detail tabel
desc = ns.describe_table(DescribeTableRequest(id=table_id))

# Bangun data uji
data = pa.table({
    "f0": pa.array([1, 2, 3], type=pa.int64()),
    "f1": pa.array(["value-1", "value-2", "value-3"], type=pa.string()),
})

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

# Baca dan verifikasi
dataset = lance.dataset(desc.location, storage_options=desc.storage_options)
print(dataset.to_table())

Lihat namespace dan tabel

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

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

# Daftar semua namespace
namespaces = ns.list_namespaces(ListNamespacesRequest(id=[]))
print(namespaces)

# Dapatkan detail namespace
namespace = ns.describe_namespace(DescribeNamespaceRequest(id=[DATABASE]))
print(namespace)

# Daftar semua tabel di database
tables = ns.list_tables(ListTablesRequest(id=[DATABASE]))
print(tables)

# Dapatkan detail tabel
table = ns.describe_table(DescribeTableRequest(id=[DATABASE, TABLE]))
print(table.location)
print(table.properties)
print(table.storage_options)

Utilitas serialisasi data

Antarmuka create_table pada modul lance_namespace memerlukan tabel Arrow yang diserialisasi sebagai aliran byte IPC (Inter-Process Communication).

import pyarrow as pa


def arrow_table_to_ipc_bytes(table: pa.Table) -> bytes:
    """Konversi tabel PyArrow ke aliran byte IPC"""
    sink = pa.BufferOutputStream()
    with pa.ipc.new_stream(sink, table.schema) as writer:
        writer.write_table(table)
    return sink.getvalue().to_pybytes()

Bangun data uji dari skema tabel DLF

Ambil informasi bidang dari skema tabel DLF untuk membangun data uji, alih-alih hardcoding nama kolom dan tipe datanya.

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


def sample_value(field_type: str, row: int):
    """Hasilkan nilai sampel berdasarkan tipe bidang"""
    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:
    """Bangun data sampel dari skema tabel 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,
)