All Products
Search
Document Center

Data Lake Formation:Access DLF with PyIceberg

Last Updated:Jul 09, 2026

This topic describes how to install PyIceberg in an EMR on ECS cluster and configure it to access DLF using the Iceberg REST protocol.

Prerequisites and installation

Install pyiceberg-dlf

pyiceberg-dlf is a DLF-compatible PyIceberg distribution published to PyPI. It includes REST sigv4 fixes, vended OSS storage credentials, and automatic credential refresh before expiry.

Requirements: Python 3.10 or later.

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]"
Note
  • Package name differs from import name: The package name is pyiceberg-dlf, but the import name remains pyiceberg.

  • pyiceberg-dlf and the official pyiceberg cannot coexist: Both packages provide import pyiceberg. Installing one silently overwrites the other. Use a dedicated virtual environment, or run pip uninstall -y pyiceberg before installing.

  • rest-sigv4 is required: This extra installs boto3 and other signing dependencies. Without it, load_catalog() raises ModuleNotFoundError: No module named 'boto3'.

  • pandas is optional: The example uses scan.to_pandas(), so include the pandas extra if you need DataFrame output.

  • pyarrow must be below 22: pyiceberg-dlf[pyarrow] pins pyarrow to 21.x. pyarrow 22 and later cause an aws-chunked encoding is not supported error when uploading to OSS.

Code example

The following Python script demonstrates how to connect to a catalog, create a table, write data, read data, and then delete the table.

Replace the placeholders in the script with your specific credentials and configuration.

  • ${regionId}: The region of your DLF service, such as cn-hangzhou. For more information, see Iceberg REST service endpoints.

  • ${catalogName}: The name of your DLF catalog.

  • ${accessKey}: Your AccessKey ID.

  • ${accessKeySecret}: Your AccessKey Secret.

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.")

Example output:

(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#