pyiceberg-core

import os import shutil import pyarrow as pa from pyiceberg.catalog import load_catalog from pyiceberg.schema import Schema from pyiceberg.types import NestedField, StringType, LongType, IntegerType # Define a temporary warehouse directory WAREHOUSE_PATH = "/tmp/pyiceberg_warehouse" CATALOG_DB_PATH = os.path.join(WAREHOUSE_PATH, "pyiceberg_catalog.db") # Clean up previous run if exists if os.path.exists(WAREHOUSE_PATH): shutil.rmtree(WAREHOUSE_PATH) os.makedirs(WAREHOUSE_PATH, exist_ok=True) # Configure and load a local SQL catalog catalog = load_catalog( "default", type="sql", uri=f"sqlite:///{CATALOG_DB_PATH}", warehouse=f"file://{WAREHOUSE_PATH}" ) # Create a namespace (database) NAMESPACE = "my_namespace" catalog.create_namespace(NAMESPACE, properties={"comment": "My first Iceberg namespace"}) print(f"Created namespace: {NAMESPACE}") # Define a schema for the Iceberg table schema = Schema( NestedField(1, "id", LongType(), required=True), NestedField(2, "name", StringType()), NestedField(3, "age", IntegerType()) ) # Create an Iceberg table TABLE_NAME = "my_table" table = catalog.create_table(f"{NAMESPACE}.{TABLE_NAME}", schema, properties={ "format-version": "2", "write.parquet.compression-codec": "zstd" }) print(f"Created table: {table.name}") # Prepare data with PyArrow data = pa.table({ "id": [1, 2, 3], "name": ["Alice", "Bob", "Charlie"], "age": [30, 24, 35] }) # Append data to the table table.append(data) print("Appended data to the table.") # Read data from the table read_df = table.scan().to_arrow() print("\nData read from Iceberg table:") print(read_df.to_pandas()) # Clean up shutil.rmtree(WAREHOUSE_PATH) print(f"Cleaned up warehouse at {WAREHOUSE_PATH}")