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]"Package name differs from import name: The package name is
pyiceberg-dlf, but the import name remainspyiceberg.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 runpip uninstall -y pyicebergbefore installing.rest-sigv4 is required: This extra installs boto3 and other signing dependencies. Without it,
load_catalog()raisesModuleNotFoundError: No module named 'boto3'.pandas is optional: The example uses
scan.to_pandas(), so include thepandasextra if you need DataFrame output.pyarrow must be below 22:
pyiceberg-dlf[pyarrow]pins pyarrow to 21.x. pyarrow 22 and later cause anaws-chunked encoding is not supportederror 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 ascn-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#