init

Author: millerjp

examples/bulk_operations/bulk_operations/bulk_operator.py

@@ -333,21 +333,6 @@ async def export_by_token_ranges(
 
         stats.end_time = time.time()
 
-    async def export_to_iceberg(
-        self,
-        source_keyspace: str,
-        source_table: str,
-        iceberg_warehouse_path: str,
-        iceberg_table: str,
-        partition_by: list[str] | None = None,
-        split_count: int | None = None,
-        batch_size: int = 10000,
-        progress_callback: Callable[[BulkOperationStats], None] | None = None,
-    ) -> BulkOperationStats:
-        """Export Cassandra table to Iceberg format."""
-        # This will be implemented when we add Iceberg integration
-        raise NotImplementedError("Iceberg export will be implemented in next phase")
-
     async def import_from_iceberg(
         self,
         iceberg_warehouse_path: str,
@@ -519,3 +504,69 @@ async def export_to_parquet(
             parallelism=parallelism,
             progress_callback=progress_callback,
         )
+
+    async def export_to_iceberg(
+        self,
+        keyspace: str,
+        table: str,
+        namespace: str | None = None,
+        table_name: str | None = None,
+        catalog: Any | None = None,
+        catalog_config: dict[str, Any] | None = None,
+        warehouse_path: str | Path | None = None,
+        partition_spec: Any | None = None,
+        table_properties: dict[str, str] | None = None,
+        compression: str = "snappy",
+        row_group_size: int = 100000,
+        columns: list[str] | None = None,
+        split_count: int | None = None,
+        parallelism: int | None = None,
+        progress_callback: Any | None = None,
+    ) -> Any:
+        """Export table data to Apache Iceberg format.
+
+        This enables modern data lakehouse features like ACID transactions,
+        time travel, and schema evolution.
+
+        Args:
+            keyspace: Cassandra keyspace to export from
+            table: Cassandra table to export
+            namespace: Iceberg namespace (default: keyspace name)
+            table_name: Iceberg table name (default: Cassandra table name)
+            catalog: Pre-configured Iceberg catalog (optional)
+            catalog_config: Custom catalog configuration (optional)
+            warehouse_path: Path to Iceberg warehouse (for filesystem catalog)
+            partition_spec: Iceberg partition specification
+            table_properties: Additional Iceberg table properties
+            compression: Parquet compression (default: snappy)
+            row_group_size: Rows per Parquet file (default: 100000)
+            columns: Columns to export (default: all)
+            split_count: Number of token range splits
+            parallelism: Max concurrent operations
+            progress_callback: Progress callback function
+
+        Returns:
+            ExportProgress with Iceberg metadata
+        """
+        from .iceberg import IcebergExporter
+
+        exporter = IcebergExporter(
+            self,
+            catalog=catalog,
+            catalog_config=catalog_config,
+            warehouse_path=warehouse_path,
+            compression=compression,
+            row_group_size=row_group_size,
+        )
+        return await exporter.export(
+            keyspace=keyspace,
+            table=table,
+            namespace=namespace,
+            table_name=table_name,
+            partition_spec=partition_spec,
+            table_properties=table_properties,
+            columns=columns,
+            split_count=split_count,
+            parallelism=parallelism,
+            progress_callback=progress_callback,
+        )

examples/bulk_operations/bulk_operations/iceberg/__init__.py

@@ -0,0 +1,15 @@
+"""Apache Iceberg integration for Cassandra bulk operations.
+
+This module provides functionality to export Cassandra data to Apache Iceberg tables,
+enabling modern data lakehouse capabilities including:
+- ACID transactions
+- Schema evolution
+- Time travel
+- Hidden partitioning
+- Efficient analytics
+"""
+
+from bulk_operations.iceberg.exporter import IcebergExporter
+from bulk_operations.iceberg.schema_mapper import CassandraToIcebergSchemaMapper
+
+__all__ = ["IcebergExporter", "CassandraToIcebergSchemaMapper"]

examples/bulk_operations/bulk_operations/iceberg/catalog.py

@@ -0,0 +1,81 @@
+"""Iceberg catalog configuration for filesystem-based tables."""
+
+from pathlib import Path
+from typing import Any
+
+from pyiceberg.catalog import Catalog, load_catalog
+from pyiceberg.catalog.sql import SqlCatalog
+
+
+def create_filesystem_catalog(
+    name: str = "cassandra_export",
+    warehouse_path: str | Path | None = None,
+) -> Catalog:
+    """Create a filesystem-based Iceberg catalog.
+
+    What this does:
+    --------------
+    1. Creates a local filesystem catalog using SQLite
+    2. Stores table metadata in SQLite database
+    3. Stores actual data files in warehouse directory
+    4. No external dependencies (S3, Hive, etc.)
+
+    Why this matters:
+    ----------------
+    - Simple setup for development and testing
+    - No cloud dependencies
+    - Easy to inspect and debug
+    - Can be migrated to production catalogs later
+
+    Args:
+        name: Catalog name
+        warehouse_path: Path to warehouse directory (default: ./iceberg_warehouse)
+
+    Returns:
+        Iceberg catalog instance
+    """
+    if warehouse_path is None:
+        warehouse_path = Path.cwd() / "iceberg_warehouse"
+    else:
+        warehouse_path = Path(warehouse_path)
+
+    # Create warehouse directory if it doesn't exist
+    warehouse_path.mkdir(parents=True, exist_ok=True)
+
+    # SQLite catalog configuration
+    catalog_config = {
+        "type": "sql",
+        "uri": f"sqlite:///{warehouse_path / 'catalog.db'}",
+        "warehouse": str(warehouse_path),
+    }
+
+    # Create catalog
+    catalog = SqlCatalog(name, **catalog_config)
+
+    return catalog
+
+
+def get_or_create_catalog(
+    catalog_name: str = "cassandra_export",
+    warehouse_path: str | Path | None = None,
+    config: dict[str, Any] | None = None,
+) -> Catalog:
+    """Get existing catalog or create a new one.
+
+    This allows for custom catalog configurations while providing
+    sensible defaults for filesystem-based catalogs.
+
+    Args:
+        catalog_name: Name of the catalog
+        warehouse_path: Path to warehouse (for filesystem catalogs)
+        config: Custom catalog configuration (overrides defaults)
+
+    Returns:
+        Iceberg catalog instance
+    """
+    if config is not None:
+        # Use custom configuration
+        return load_catalog(catalog_name, **config)
+    else:
+        # Use filesystem catalog
+        return create_filesystem_catalog(catalog_name, warehouse_path)