Spark Metadata Extraction

geoembed.metadata.spark

Spark-based distributed metadata extraction for GeoTIFF imagery.

Scans a Unity Catalog Volume (or any path) for GeoTIFF files, extracts spatial and temporal metadata from headers + XML sidecars in parallel across Spark workers, and writes results to a Delta table.

This is typically the FIRST stage of the pipeline — it creates the metadata table that subsequent stages (COG conversion, chipping) read from.

MetadataExtractionConfig(volume_path, output_table, glob_pattern='*.tif', recursive=True) dataclass

Configuration for Spark-based metadata extraction.

Attributes:

Name	Type	Description
volume_path	str	Path to scan for GeoTIFFs (Unity Catalog Volume or local).
output_table	str	Delta table to write metadata to.
glob_pattern	str	File pattern to match (default: "*.tif").
recursive	bool	Scan subdirectories recursively.

MetadataExtractionWorker

Spark mapInPandas worker for distributed metadata extraction.

Initialises GDAL optimisations and metadata readers once per partition, then processes batches of file paths. Extracts: - Spatial bounds and geometry (from GeoTIFF header via rasterio) - Temporal extent, EPSG, and provider (from XML sidecar if present)

The worker is serialised to executors — keep it lightweight at init time.

SparkMetadataExtractor(spark, config)

Distributed metadata extraction pipeline for Databricks.

Scans a Unity Catalog Volume for GeoTIFFs, distributes header reading across Spark workers, and writes a Delta metadata table.

Usage

from geoembed.metadata.spark import SparkMetadataExtractor, MetadataExtractionConfig

config = MetadataExtractionConfig( volume_path="/Volumes/catalog/schema/raw_imagery", output_table="catalog.schema.imagery_metadata", ) extractor = SparkMetadataExtractor(spark, config) extractor.run()

Parameters:

Name	Type	Description	Default
spark	Any	Active SparkSession.	required
config	MetadataExtractionConfig	Extraction configuration.	required

Source code in src/geoembed/metadata/spark.py

def __init__(self, spark: Any, config: MetadataExtractionConfig):
    """
    Args:
        spark: Active SparkSession.
        config: Extraction configuration.
    """
    self.spark = spark
    self.config = config

run()

Execute the metadata extraction pipeline.

Scans the volume, distributes extraction, writes to Delta with native geometry.

Source code in src/geoembed/metadata/spark.py

def run(self) -> None:
    """
    Execute the metadata extraction pipeline.

    Scans the volume, distributes extraction, writes to Delta with native geometry.
    """
    start_time = time.monotonic()
    quoted_table = quote_table_name(self.config.output_table)

    print(f"[SparkMetadataExtractor] Scanning {self.config.volume_path}...")
    path_df = (
        self.spark.read.format("binaryFile")
        .option("pathGlobFilter", self.config.glob_pattern)
        .option("recursiveFileLookup", str(self.config.recursive).lower())
        .load(self.config.volume_path)
        .select("path")
    )

    total_files = path_df.count()
    print(f"[SparkMetadataExtractor] Found {total_files} files")

    if total_files == 0:
        print("[SparkMetadataExtractor] No files found. Exiting.")
        return

    worker = MetadataExtractionWorker()
    result_df = path_df.mapInPandas(worker, schema=self.OUTPUT_SCHEMA)

    result_df = self._add_native_geometry(result_df)

    (
        result_df.write.format("delta")
        .mode("overwrite")
        .option("overwriteSchema", "true")
        .saveAsTable(quoted_table)
    )

    self.spark.sql(f"ALTER TABLE {quoted_table} CLUSTER BY (id)")

    elapsed = time.monotonic() - start_time
    row_count = self.spark.read.table(quoted_table).count()
    error_count = self.spark.read.table(quoted_table).filter("id LIKE 'ERROR_%'").count()

    print(
        f"[SparkMetadataExtractor] Complete: {row_count} rows "
        f"({error_count} errors) in {elapsed:.1f}s"
    )
    print(f"[SparkMetadataExtractor] Output: {self.config.output_table}")