How to read/write Iceberg table from/to ADLS gen 2 container with PyIceberg

[HungYangChang]

Question

I would like to read/write the Iceberg table from/to ADLS gen2 with PyIceberg

Background:

• I know how to use Spark to read/write the file, and I has successfully done it locally
• The code I use to upload Iceberg table with nessie server

ACCOUNT_NAME = "example-events"  # Storage account name
CONTAINER_NAME = "example-events-iceberg-debug-local"  # Container name
WAREHOUSE_PATH = f"abfs://{CONTAINER_NAME}@{ACCOUNT_NAME}.dfs.core.windows.net/"

spark = SparkSession.builder \
            .config("spark.jars", 'jars/bundle-2.17.178.jar,'
                                  'jars/iceberg-spark-runtime-3.5_2.12-1.5.2.jar,'
                                  'jars/nessie-spark-extensions-3.5_2.12-0.101.3.jar,'
                                  'jars/url-connection-client-2.17.178.jar,'
                                  'jars/postgresql-42.5.0.jar,'
                                  'jars/azure-storage-blob-12.20.0.jar,'
                                  'jars/azure-identity-1.8.1.jar,'
                                  "jars/hadoop-common-3.3.5.jar,jars/hadoop-client-3.3.5.jar,jars/hadoop-azure-3.3.5.jar,"
                                  'jars/hadoop-azure-datalake-3.3.5.jar') \
            .config("spark.hadoop.io.native.lib.available", "false") \
            .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions,"
                                            "org.projectnessie.spark.extensions.NessieSparkSessionExtensions") \
            .config("spark.sql.catalog.nessie.uri", "http://localhost:19120/api/v1") \
            .config("spark.sql.catalog.nessie.ref", ref) \
            .config("spark.sql.catalog.nessie.catalog-impl", "org.apache.iceberg.nessie.NessieCatalog") \
            .config("spark.sql.catalog.nessie.warehouse", WAREHOUSE_PATH) \
            .config("spark.sql.catalog.nessie", "org.apache.iceberg.spark.SparkCatalog") \
            .config("spark.driver.extraJavaOptions", f"-Dhadoop.home.dir={hadoop_home}") \
            .config("spark.executor.extraJavaOptions", f"-Dhadoop.home.dir={hadoop_home}") \
            .config("spark.hadoop.fs.azure.account.auth.type.xuievents.dfs.core.windows.net", "OAuth") \
            .config("spark.hadoop.fs.azure.account.oauth.provider.type.xuievents.dfs.core.windows.net",
                    "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider") \
            .config("spark.hadoop.fs.azure.account.oauth2.client.id.xuievents.dfs.core.windows.net", AZURE_CLIENT_ID) \
            .config("spark.hadoop.fs.azure.account.oauth2.client.secret.xuievents.dfs.core.windows.net", AZURE_CLIENT_SECRET) \
            .config("spark.hadoop.fs.azure.account.oauth2.client.endpoint.xuievents.dfs.core.windows.net",
                    f"https://login.microsoftonline.com/{AZURE_TENANT_ID}/oauth2/token") \
            .getOrCreate()

        # Create namespace
        spark.sql(f"CREATE NAMESPACE IF NOT EXISTS nessie.{NAME_SPACE}")
        print(f"Namespace '{NAME_SPACE}' created successfully!")

        # Create a table
        spark.sql(f"CREATE TABLE nessie.{NAME_SPACE}.names (name STRING) USING iceberg")
        print("Table 'names' created successfully!")

        # Insert data
        spark.sql(f"INSERT INTO nessie.{NAME_SPACE}.names VALUES ('Alex Merced'), ('Dipankar Mazumdar'), ('Jason Hughes')")
        print("Data inserted successfully!")

I can confirm it has successfully uploaded to local nessie server http://localhost:19120 and data is stored in ADLS gen 2 container

Nessie server:

ADLS gen 2:

However, I still cannot figure out how to read/write the Iceberg table from/to ADLS gen2 with PyIceberg

I have reviewed https://py.iceberg.apache.org/#getting-started-with-pyiceberg but still have no luck

[kevinjqliu]

I still cannot figure out how to read/write the Iceberg table from/to ADLS gen2 with PyIceberg

in pyiceberg, you would need to configure both the catalog and fileio

[HungYangChang]

Hi @kevinjqliu

Thanks for the hint
I still have some issues set up the fileio for ADLS gen2 storage
This is my script:

STORAGE_ACCOUNT = "xuievents"
CONTAINER_NAME = "lakekeeper-test-storage"
credential = ClientSecretCredential(
    tenant_id=AZURE_TENANT_ID,
    client_id=AZURE_CLIENT_ID,
    client_secret=AZURE_CLIENT_SECRET
)

# solution 1
# error: TypeError: FsspecFileIO.__init__() got an unexpected keyword argument 'fs'
fs_adl = fsspec.filesystem(
    "abfs",
    account_name=STORAGE_ACCOUNT,
    credential=credential, 
)
file_io = FsspecFileIO(fs=fs_adl)

# solution 2
# error: Must provide either a connection_string or account_name with credentials!!
file_io = FsspecFileIO(properties={"account_name": STORAGE_ACCOUNT, "credential": credential})

# soultion 3
# error: Must provide either a connection_string or account_name with credentials!!
file_io = FsspecFileIO("abfs://lakekeeper-test-storage@xuievents.dfs.core.windows.net/")


# solution 4
# error: Must provide either a connection_string or account_name with credentials!!
azure_storage_options = {
    "account_name": STORAGE_ACCOUNT,
    "tenant_id": AZURE_TENANT_ID,
    "client_id": AZURE_CLIENT_ID,
    "client_secret": AZURE_CLIENT_SECRET
}

# Initialize FsspecFileIO with the Azure storage options
file_io = FsspecFileIO(
    properties={
        "protocol": "abfs",
        "storage_options": azure_storage_options
    }
)

CATALOG_URL = "http://localhost:8181/catalog"
WAREHOUSE_PATH = "test"
catalog = RestCatalog(
    name="test_catalog",
    uri=CATALOG_URL,
    warehouse=WAREHOUSE_PATH,
    token="dummy",
    properties={
        "azure.account-name": STORAGE_ACCOUNT,
        "azure.container": CONTAINER_NAME,
        "azure.endpoint": "https://xuievents.dfs.core.windows.net",
        "azure.client-id": AZURE_CLIENT_ID,
        "azure.tenant-id": AZURE_TENANT_ID,
        "azure.client-secret": AZURE_CLIENT_SECRET,
        "io-impl": "pyiceberg.io.fsspec.FsspecFileIO",
    }
)
catalog.file_io = file_io

Thanks for your answer in advance

[HungYangChang]

I found out actually it's not the file_io issue

I am able to solved it
My new code is working now (test against lakekeeper)

set up env variable (this solves this error: Must provide either a connection_string or account_name with credentials!!  ):

AZURE_STORAGE_CLIENT_ID = os.environ.get("AZURE_STORAGE_CLIENT_ID")
AZURE_STORAGE_CLIENT_SECRET = os.environ.get("AZURE_STORAGE_CLIENT_SECRET")
AZURE_STORAGE_TENANT_ID = os.environ.get("AZURE_STORAGE_TENANT_ID")
AZURE_STORAGE_ACCOUNT_NAME = os.environ.get("AZURE_STORAGE_ACCOUNT_NAME")
credential = ClientSecretCredential(
    tenant_id=AZURE_STORAGE_TENANT_ID,
    client_id=AZURE_STORAGE_CLIENT_ID,
    client_secret=AZURE_STORAGE_CLIENT_SECRET
)
file_io = FsspecFileIO(properties={"account_name": AZURE_STORAGE_ACCOUNT_NAME, "credential": credential})

catalog = RestCatalog(
    name="test_catalog",
    uri=CATALOG_URL,
    warehouse=WAREHOUSE_PATH,
    token="dummy",
    properties={
        # either adlfs or adls can be used
        "adlfs.account-name": AZURE_STORAGE_ACCOUNT_NAME,
        "adlfs.container": CONTAINER_NAME,
        "adlfs.client-id": AZURE_STORAGE_CLIENT_ID,
        "adlfs.tenant-id": AZURE_STORAGE_TENANT_ID,
        "adlfs.client-secret": AZURE_STORAGE_CLIENT_SECRET,  # ✅ Use Service Principal
        "client_secret": AZURE_STORAGE_CLIENT_SECRET,
        "client_id": AZURE_STORAGE_CLIENT_ID,
        "tenant_id": AZURE_STORAGE_TENANT_ID,
        "io-impl": "pyiceberg.io.fsspec.FsspecFileIO",
    }
)

catalog.file_io = file_io

[martinseifertprojuventute]

Hey @HungYangChang

Did you try to table.scan().to_pandas() with that? I still fail at that step :/