COG Conversion

geoembed.images.cog

Cloud Optimised GeoTIFF (COG) conversion pipeline.

Converts standard GeoTIFFs to COGs with internal tiling, compression, and overviews for efficient cloud-based partial reads (HTTP range requests).

Includes a Spark-based pipeline for distributed conversion on Databricks.

FileSystemAdapter

Bases: ABC

Abstract filesystem operations for testability.

LocalFileSystem

Bases: FileSystemAdapter

Concrete filesystem adapter for local/mounted paths.

ImageConverter

Bases: ABC

Abstract interface for image format conversion.

RioCogConverter(compression='deflate', blocksize=512, overview_resampling='nearest')

Bases: ImageConverter

Converts GeoTIFFs to COGs optimised for high-throughput GPU inference.

Based on Microsoft's COG throughput findings (Zaytar et al., 2025):

Compression choice depends on storage type: - LOCAL STORAGE: "none" (uncompressed) is 1.3-1.6x faster because CPU decompression is the bottleneck when reading from fast local disk/SSD. - REMOTE/CLOUD STORAGE: "lerc_zstd" is optimal — balances network transfer reduction (good compression ratio) with fast CPU decompression. - "deflate" is a safe middle ground for mixed local/remote workflows.

Tiling: - 512x512 internal tiles enable tile-aligned reads. A 224px chip falls entirely within one tile, requiring exactly one decompression. - Misaligned reads force decompression of up to 4 adjacent tiles.

Overviews: - Embedded reduced-resolution pyramids for fast wide-area previews.

Parameters:

Name	Type	Description	Default
compression	str	Compression codec. Recommendations: - "none": Fastest local reads (no decompression overhead). Largest files. - "deflate": Good default. Moderate compression + decode speed. - "lerc_zstd": Best for remote/cloud (high compression, fast decode). - "lzw": Legacy option, similar to deflate.	'deflate'
blocksize	int	Internal tile size. Must be 256 or 512 for tile-aligned reads. 512 recommended — a 224px chip fits within one tile.	512
overview_resampling	str	Resampling for overviews. "nearest" preserves spectral accuracy; "average" for smoother visual previews.	'nearest'

Source code in src/geoembed/images/cog.py

def __init__(
    self,
    compression: str = "deflate",
    blocksize: int = 512,
    overview_resampling: str = "nearest",
):
    """
    Args:
        compression: Compression codec. Recommendations:
            - "none": Fastest local reads (no decompression overhead). Largest files.
            - "deflate": Good default. Moderate compression + decode speed.
            - "lerc_zstd": Best for remote/cloud (high compression, fast decode).
            - "lzw": Legacy option, similar to deflate.
        blocksize: Internal tile size. Must be 256 or 512 for tile-aligned reads.
            512 recommended — a 224px chip fits within one tile.
        overview_resampling: Resampling for overviews. "nearest" preserves
            spectral accuracy; "average" for smoother visual previews.
    """
    self.compression = compression
    self.blocksize = blocksize
    self.overview_resampling = overview_resampling

convert(src, dst)

Convert a GeoTIFF to a GPU-cloud-friendly COG.

Generates a COG with internal 512x512 tiling, compression, and overviews.

Source code in src/geoembed/images/cog.py

def convert(self, src: str, dst: str) -> None:
    """
    Convert a GeoTIFF to a GPU-cloud-friendly COG.

    Generates a COG with internal 512x512 tiling, compression, and overviews.
    """
    from rio_cogeo.cogeo import cog_translate
    from rio_cogeo.profiles import cog_profiles

    output_profile = cog_profiles.get(self.compression)
    output_profile.update({
        "blockxsize": self.blocksize,
        "blockysize": self.blocksize,
        "predictor": 2,
    })

    config = {
        # Use 2 threads per file — avoids thread starvation when multiple
        # Spark tasks share the same executor node.
        "GDAL_NUM_THREADS": "2",
        "GDAL_TIFF_INTERNAL_MASK": True,
        "GDAL_TIFF_OVR_BLOCKSIZE": self.blocksize,
    }

    cog_translate(
        src,
        dst,
        output_profile,
        config=config,
        overview_resampling=self.overview_resampling,  # pyrefly: ignore
        use_cog_driver=True,
    )

validate(path)

Validate that a file is a valid COG.

Source code in src/geoembed/images/cog.py

def validate(self, path: str) -> bool:
    """Validate that a file is a valid COG."""
    from rio_cogeo.cogeo import cog_validate

    is_valid, _, _ = cog_validate(path)
    return is_valid

CogConfig(input_table, output_table, dst_root, compression='deflate', blocksize=512, files_per_task=10) dataclass

Configuration for the COG conversion pipeline.

Attributes:

Name	Type	Description
input_table	str	Source table with image metadata (path column required).
output_table	str	Table to log conversion results.
dst_root	str	Destination directory for COG files.
compression	str	Compression codec.
blocksize	int	Internal tile size.
files_per_task	int	Files to process per Spark task.

CogTaskHandler(converter=None, fs=None, dst_root='/tmp/cog_output')

Handles COG conversion for a batch of files.

Writes to a unique local temp directory (NVMe /local_disk0 on Databricks), then copies the validated COG to the UC Volume destination.

Source code in src/geoembed/images/cog.py

def __init__(
    self,
    converter: ImageConverter | None = None,
    fs: FileSystemAdapter | None = None,
    dst_root: str = "/tmp/cog_output",  # noqa: S108
):
    self.converter = converter or RioCogConverter()
    self.fs = fs or LocalFileSystem()
    self.dst_root = dst_root

process_file(src_path)

Convert a single file to COG.

Uses a unique temp directory per file to avoid collisions when multiple Spark tasks run concurrently on the same executor.

Parameters:

Name	Type	Description	Default
src_path	str	Path to the source GeoTIFF.	required

Returns:

Type	Description
dict[str, Any]	Dict with keys: src, status, error.

Source code in src/geoembed/images/cog.py

def process_file(self, src_path: str) -> dict[str, Any]:
    """
    Convert a single file to COG.

    Uses a unique temp directory per file to avoid collisions when multiple
    Spark tasks run concurrently on the same executor.

    Args:
        src_path: Path to the source GeoTIFF.

    Returns:
        Dict with keys: src, status, error.
    """
    filename = os.path.basename(src_path)
    # Use /local_disk0 on Databricks (fast NVMe) with unique dir per file
    temp_dir = f"/local_disk0/tmp/cog_{uuid.uuid4().hex[:8]}"
    temp_path = os.path.join(temp_dir, filename)
    final_path = os.path.join(self.dst_root, filename)

    try:
        os.makedirs(temp_dir, exist_ok=True)
        self.fs.makedirs(self.dst_root)

        # Convert to local temp (fast, NVMe)
        self.converter.convert(src_path, temp_path)

        # Validate locally (fast)
        if not self.converter.validate(temp_path):
            return {"src": src_path, "status": "invalid", "error": "COG validation failed"}

        # Copy validated COG to UC Volume (FUSE upload)
        self.fs.copy(temp_path, final_path)
    except Exception as e:
        return {"src": src_path, "status": "error", "error": str(e)}
    else:
        return {"src": src_path, "status": "success", "error": None}
    finally:
        # Always clean up local temp
        shutil.rmtree(temp_dir, ignore_errors=True)

SparkCogWorker(config)

Spark mapInPandas worker for distributed COG conversion.

Pickled to Spark executors. Each worker converts a batch of GeoTIFFs to COGs.

Source code in src/geoembed/images/cog.py

def __init__(self, config: CogConfig):
    self.config = config

SparkCogPipeline(spark, config)

Distributed COG conversion pipeline for Databricks.

Scans a volume for GeoTIFFs, distributes conversion across Spark workers using mapInPandas, and writes COGs to a destination volume.

Usage

from geoembed.images.cog import SparkCogPipeline, CogConfig

config = CogConfig( input_table="catalog.schema.imagery_metadata", output_table="catalog.schema.cog_log", dst_root="/Volumes/catalog/data/processed/COG", compression="deflate", ) pipeline = SparkCogPipeline(spark, config) pipeline.run()

Parameters:

Name	Type	Description	Default
spark	Any	Active SparkSession.	required
config	CogConfig	COG conversion configuration.	required

Source code in src/geoembed/images/cog.py

def __init__(self, spark: Any, config: CogConfig):
    """
    Args:
        spark: Active SparkSession.
        config: COG conversion configuration.
    """
    self.spark = spark
    self.config = config

run()

Run COG conversion from an existing metadata table.

Reads image_path column from config.input_table.

Source code in src/geoembed/images/cog.py

def run(self) -> None:
    """
    Run COG conversion from an existing metadata table.

    Reads image_path column from config.input_table.
    """
    from pyspark.sql.functions import col

    quoted_input = quote_table_name(self.config.input_table)
    print(f"[SparkCogPipeline] Reading source paths from {self.config.input_table}")
    df = self.spark.read.table(quoted_input).select(col("image_path").alias("path"))
    self._process(df)

run_from_volume(volume_path, glob_pattern='*.tif')

Run COG conversion by scanning a Unity Catalog Volume for GeoTIFFs.

Parameters:

Name	Type	Description	Default
volume_path	str	Path to the UC Volume.	required
glob_pattern	str	File glob pattern (default: "*.tif").	'*.tif'

Source code in src/geoembed/images/cog.py

def run_from_volume(self, volume_path: str, glob_pattern: str = "*.tif") -> None:
    """
    Run COG conversion by scanning a Unity Catalog Volume for GeoTIFFs.

    Args:
        volume_path: Path to the UC Volume.
        glob_pattern: File glob pattern (default: "*.tif").
    """
    print(f"[SparkCogPipeline] Scanning {volume_path} for {glob_pattern}")
    df = (
        self.spark.read.format("binaryFile")
        .option("pathGlobFilter", glob_pattern)
        .option("recursiveFileLookup", "true")
        .load(volume_path)
        .select("path")
    )
    self._process(df)