Accéder à DLF avec PyIceberg - Data Lake Formation - Centre de documentation Alibaba Cloud

Cette rubrique explique comment installer PyIceberg dans un cluster EMR on ECS et le configurer pour accéder à DLF via le protocole REST Iceberg.

Prérequis et installation

Installer pyiceberg-dlf

pyiceberg-dlf est une distribution PyIceberg compatible avec DLF, publiée sur PyPI. Elle intègre les correctifs REST sigv4, la fourniture d'identifiants de stockage OSS et l'actualisation automatique des identifiants avant leur expiration.

Prérequis : Python 3.10 ou version ultérieure.

python3 -m venv venv
source venv/bin/activate
pip install -U pip
# Uninstall the official package (cannot coexist with pyiceberg-dlf)
pip uninstall -y pyiceberg
# rest-sigv4 is required (installs boto3 for REST sigv4 signing)
pip install "pyiceberg-dlf[rest-sigv4,pyarrow,pandas]"

Remarque

Le nom du package diffère du nom d'importation : Le nom du package est pyiceberg-dlf, mais le nom d'importation reste pyiceberg.
Incompatibilité entre pyiceberg-dlf et la version officielle de pyiceberg : Les deux packages proposent import pyiceberg. L'installation de l'un écrase silencieusement l'autre. Utilisez un environnement virtuel dédié ou exécutez pip uninstall -y pyiceberg avant l'installation.
rest-sigv4 est requis : Cette option supplémentaire installe boto3 et d'autres dépendances de signature. Sans elle, load_catalog() génère l'erreur ModuleNotFoundError: No module named 'boto3'.
pandas est facultatif : L'exemple utilise scan.to_pandas(). Incluez l'option supplémentaire pandas si vous avez besoin d'une sortie au format DataFrame.
La version de pyarrow doit être inférieure à 22 : pyiceberg-dlf[pyarrow] fixe pyarrow à la version 21.x. Les versions 22 et ultérieures de pyarrow provoquent une erreur aws-chunked encoding is not supported lors du chargement vers OSS.

Exemple de code

Le script Python suivant illustre la connexion à un catalogue, la création d'une table, l'écriture et la lecture de données, puis la suppression de la table.

Remplacez les espaces réservés du script par vos propres identifiants et votre configuration.

${regionId} : Région de votre service DLF, par exemple cn-hangzhou. Pour plus d'informations, consultez la section Points de terminaison du service REST Iceberg.
${catalogName} : Nom de votre catalogue DLF.
${accessKey} : Votre ID AccessKey.
${accessKeySecret} : Votre secret AccessKey.

import pyarrow as pa
from pyiceberg.catalog import load_catalog
from pyiceberg.exceptions import TableAlreadyExistsError, NoSuchTableError
from pyiceberg.io.pyarrow import schema_to_pyarrow
from pyiceberg.partitioning import PartitionField, PartitionSpec
from pyiceberg.schema import Schema
from pyiceberg.transforms import IdentityTransform
from pyiceberg.types import LongType, NestedField
# 1. Configure and connect to the catalog
catalog = load_catalog(
    "default",
    **{
        "type": "rest",
        "uri": "http://${regionId}-vpc.dlf.aliyuncs.com/iceberg",
        "warehouse": "${catalogName}",
        "rest.signing-name": "DlfNext",
        "rest.signing-region": "${regionId}",
        "rest.sigv4-enabled": "true",
        "client.access-key-id": "${accessKey}",
        "client.secret-access-key": "${accessKeySecret}",
        "client.region": "${regionId}",
    },
)
# ---------------------------------------------------
# 2. Define the table schema and metadata
# ---------------------------------------------------
TEST_TABLE_SCHEMA = Schema(
    NestedField(1, "x", LongType(), required=True),
    NestedField(2, "y", LongType(), doc="comment", required=True),
    NestedField(3, "z", LongType(), required=True),
)
TEST_TABLE_IDENTIFIER = ("default", "my_table")
TEST_TABLE_PARTITION_SPEC = PartitionSpec(
    PartitionField(name="x", transform=IdentityTransform(), source_id=1, field_id=1000)
)
TEST_TABLE_PROPERTIES = {"read.split.target.size": "134217728"}  # 128MB
# ---------------------------------------------------
# 3. Run the test procedure
# ---------------------------------------------------
# Drop the old table if it exists
try:
    catalog.drop_table(identifier=TEST_TABLE_IDENTIFIER)
    print("Existing table dropped.")
except NoSuchTableError:
    print("No existing table to drop.")
# Create a new table
try:
    catalog.create_table(
        identifier=TEST_TABLE_IDENTIFIER,
        schema=TEST_TABLE_SCHEMA,
        partition_spec=TEST_TABLE_PARTITION_SPEC,
        properties=TEST_TABLE_PROPERTIES,
    )
    print("Table created.")
except TableAlreadyExistsError:
    print("Table already exists, will append data.")
# Load the table
table = catalog.load_table(identifier=TEST_TABLE_IDENTIFIER)
print(f"Loaded table: {table}")
# Build a PyArrow table
arrow_schema = schema_to_pyarrow(table.schema())
data = pa.Table.from_pydict(
    {
        "x": [1, 2, 3],
        "y": [10, 20, 30],
        "z": [100, 200, 300],
    },
    schema=arrow_schema,
)
# Write data
print(f"Inserting {data.num_rows} rows...")
table.append(data)
print("Insert finished.")
# Read and display the data
scan = table.scan()
df = scan.to_pandas()
print("First 10 rows via to_pandas():")
print(df.head(10))
# Clean up the test table
try:
    catalog.drop_table(identifier=TEST_TABLE_IDENTIFIER)
    print("Table dropped after test.")
except NoSuchTableError:
    print("Table already dropped.")

Exemple de sortie :

(myenv) root@iZbp1h4zr65vjcvz9w3080Z:~/workspace# python test.py
Existing table dropped.
Table created.
Loaded table: my_table(
    1: x: required long,
    2: y: required long (comment),
    3: z: required long
),
partition by: [x],
sort order: [],
snapshot: null
Inserting 3 rows...
Insert finished.
First 10 rows via to_pandas():
   x   y    z
0  1  10  100
1  2  20  200
2  3  30  300
Table dropped after test.
(myenv) root@iZbp1h4zr65vjcvz9w3080Z:~/workspace# pip list|grep -E "pyarrow|pyiceberg|boto3|pandas"
boto3                 1.42.15
pandas                2.3.3
pyarrow               19.0.0
pyiceberg             0.10.0.dev0
(myenv) root@iZbp1h4zr65vjcvz9w3080Z:~/workspace#