pyiceberg

import os from pyiceberg.catalog import load_catalog from pyiceberg.schema import Schema, NestedField, PrimitiveType import pyarrow as pa import datetime # Configure a local SQL catalog using environment variables or direct properties # For simplicity, we'll use a temporary local directory. warehouse_path = '/tmp/pyiceberg_warehouse' os.makedirs(warehouse_path, exist_ok=True) catalog = load_catalog( "default", type="sql", uri=f"sqlite:///{warehouse_path}/pyiceberg_catalog.db", warehouse=f"file://{warehouse_path}" ) print(f"✓ Successfully loaded catalog: {catalog.name}") # Define a simple schema schema = Schema( NestedField(1, "id", PrimitiveType.long(), required=True), NestedField(2, "name", PrimitiveType.string(), required=False), NestedField(3, "event_time", PrimitiveType.timestamp_ntz(), required=False) ) namespace = "default" table_name = "my_sample_table" # Create a namespace if it doesn't exist catalog.create_namespace_if_not_exists(namespace) print(f"✓ Ensured namespace '{namespace}' exists.") # Create a table try: table = catalog.create_table(f"{namespace}.{table_name}", schema) print(f"✓ Successfully created table: {table.identifier}") except Exception as e: print(f"Table {namespace}.{table_name} might already exist. Loading it instead. Error: {e}") table = catalog.load_table(f"{namespace}.{table_name}") # Prepare some data using PyArrow data = pa.table({ "id": [1, 2, 3], "name": ["Alice", "Bob", "Charlie"], "event_time": [datetime.datetime.now() - datetime.timedelta(days=i) for i in range(3)] }) # Append data to the table table.append(data) print(f"✓ Appended {len(data)} rows to the table.") # Read data from the table scan_result = table.scan().to_arrow() print(f"\nTotal rows read: {len(scan_result)}") print("Sample data:") print(scan_result.to_pandas()) # Clean up (optional: uncomment to drop the table and namespace) # catalog.drop_table(f"{namespace}.{table_name}") # catalog.drop_namespace(namespace) # print(f"Cleaned up table {table_name} and namespace {namespace}.")