From c7a32cf3d1a8c7360b5bb96fa6668cc321334fbb Mon Sep 17 00:00:00 2001 From: Michael Hemming Date: Thu, 16 Jul 2026 15:12:18 +1000 Subject: [PATCH 1/3] tool to analyse NetCDF schema for a prefix --- pyproject.toml | 5 + .../cli/ANALYSE_NETCDF_SCHEMA_README.md | 337 ++++++++++++++++++ src/data_index/cli/__init__.py | 1 + src/data_index/cli/analyse_netcdf_schema.py | 334 +++++++++++++++++ 4 files changed, 677 insertions(+) create mode 100644 src/data_index/cli/ANALYSE_NETCDF_SCHEMA_README.md create mode 100644 src/data_index/cli/__init__.py create mode 100644 src/data_index/cli/analyse_netcdf_schema.py diff --git a/pyproject.toml b/pyproject.toml index 367b35b..e4ae9e5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -49,6 +49,7 @@ source = "vcs" data-index = "data_index.component_testing:main" cluster-local = "data_index.cluster.run_local:main" cluster-fargate = "data_index.cluster.run_fargate:main" +analyse-netcdf-schema = "data_index.cli.analyse_netcdf_schema:app" [tool.pytest.ini_options] filterwarnings = ["ignore:Logger 'prefect.*' attempted to send logs"] @@ -82,3 +83,7 @@ analysis = [ "natsort>=8.4.0", "rich>=13", ] +cli = [ + "typer>=0.12.0", + "rich>=13.0.0", +] diff --git a/src/data_index/cli/ANALYSE_NETCDF_SCHEMA_README.md b/src/data_index/cli/ANALYSE_NETCDF_SCHEMA_README.md new file mode 100644 index 0000000..e4988a7 --- /dev/null +++ b/src/data_index/cli/ANALYSE_NETCDF_SCHEMA_README.md @@ -0,0 +1,337 @@ +# NetCDF Schema Variance Analysis + +This script analyzes the structure and variance of NetCDF datasets stored in the S3 Tables Iceberg catalog. It helps identify schema inconsistencies across collections and facilities, which is crucial for understanding data quality and compatibility. + +## Overview + +NetCDF datasets often have variant schemas, especially when: +- Datasets span multiple decades of collection +- Different facilities or projects manage data independently +- Codebase changes or hands-offs occur over time + +This script examines the `variables` column across records to quantify schema variance and identify which variable schemas are most common. + +## Features + +- **Typer CLI Framework**: Professional command-line interface with auto-generated help +- **Rich Output**: Colored, formatted output with tables and formatting +- **AWS Credential Handling**: Automatic credential detection and setup +- **Schema Normalization**: Treats schemas with identical variables (in any order) as equivalent +- **Variable Analysis**: Identifies unique and missing variables per schema +- **Flexible Output**: Save results to custom directories +- **Performance**: Column pruning and row filtering for efficient S3 queries + +## Prerequisites + +- Python 3.10+ +- `uv` package manager +- `boto3` (for AWS credential handling) +- Access to AWS S3 Tables with appropriate credentials +- AWS credentials configured via one of: + - AWS CLI: `aws configure` + - Environment variables: `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY` + - IAM role (if running on EC2 or ECS) + +## Installation + +1. Install with CLI dependencies: + ```bash + uv sync --group cli + ``` + +2. Configure AWS credentials using one of these methods: + + **Option A: AWS CLI Configuration** + ```bash + aws configure + # Enter your AWS Access Key ID and Secret Access Key when prompted + ``` + + **Option B: Environment Variables** + ```bash + export AWS_ACCESS_KEY_ID=your_access_key + export AWS_SECRET_ACCESS_KEY=your_secret_key + ``` + + **Option C: IAM Role** + If running on EC2 or ECS, the script will automatically use your IAM role credentials. + +3. Verify credentials are working: + ```bash + aws sts get-caller-identity + ``` + +## Usage + +### CLI Commands + +View help: +```bash +uv run python analyse_netcdf_schema.py --help +``` + +Analyse all data: +```bash +uv run python analyse_netcdf_schema.py +``` + +Analyse a specific S3 prefix: +```bash +uv run python analyse_netcdf_schema.py "IMOS/ANFOG/%" +``` + +Analyse with custom output directory: +```bash +uv run python analyse_netcdf_schema.py "IMOS/SOOP/SOOP-CO2%" --output-dir ./results +``` + +Other examples: +```bash +uv run python analyse_netcdf_schema.py "IMOS/SOOP/%" +uv run python analyse_netcdf_schema.py "IMOS/ANMN%" +``` + +### CLI Options + +- `S3_PREFIX` (optional): S3 prefix to analyse (e.g., `'IMOS/ANFOG/%'`, `'IMOS/SOOP/SOOP-CO2%'`) +- `--output-dir, -o`: Directory to save results (default: current directory) +- `--help`: Show help message + +The script will display results to console with a rich formatted table and save detailed analysis to a timestamped output file: +- `netcdf_schema_analysis_all_20260716_141000.txt` (for all data) +- `netcdf_schema_analysis_IMOS_ANFOG_wildcard_20260716_141000.txt` (for IMOS/ANFOG/%) + +### Programmatic Usage + +```python +from analyse_netcdf_schema import load_table, query_with_filter + +# Load the table +table = load_table() + +# Query with S3 prefix filter +df = query_with_filter( + table, + s3_prefix="IMOS/ANFOG/%", + selected_fields=("key", "variables") +) + +# Analyze variance +print(df["variables"].value_counts(sort=True)) +``` + +## Functions + +### `load_table(namespace, table_name, region, arn)` + +Loads an Iceberg table from the S3 Tables catalog with automatic AWS credential handling. + +**Parameters:** +- `namespace` (str): Iceberg namespace, default: `"data_index"` +- `table_name` (str): Table name, default: `"structured_metadata_v5"` +- `region` (str): AWS region, default: `"ap-southeast-2"` +- `arn` (str): S3 Tables bucket ARN + +**Returns:** `pyiceberg.Table` object + +### `query_with_filter(table, s3_prefix, row_filter, selected_fields)` + +Queries the table with filtering and column pruning for performance. + +**Parameters:** +- `table`: Iceberg table object +- `s3_prefix` (str, optional): S3 prefix to filter by (e.g., `"IMOS/ANFOG/%"`) +- `row_filter` (str, optional): Additional row filter predicate +- `selected_fields` (tuple, optional): Fields to retrieve (column pruning) + +**Returns:** Polars DataFrame + +### `analyze_schema_variance(df, group_by)` + +Analyzes schema variance by examining the `variables` column with normalization. + +**Key Features:** +- Schemas with the same variables in any order are treated as identical +- Tracks which variables are unique to each schema +- Identifies missing variables in each schema variant +- Identifies directory prefixes associated with each schema + +**Parameters:** +- `df`: Polars DataFrame with `variables` and `key` columns +- `group_by` (str, optional): Column to group analysis by + +**Returns:** Dict with normalized schema variants, variable-level analysis, and their associated directory prefixes + +### `save_results(results, s3_prefix)` + +Saves analysis results to a timestamped output file. + +**Parameters:** +- `results`: Analysis results dictionary from `analyze_schema_variance()` +- `s3_prefix` (str, optional): S3 prefix used for querying (used in filename) + +**Returns:** Output file path + +## Performance Tips + +1. **Use field pruning** - Select only the fields you need: + ```python + selected_fields=("key", "variables") + ``` + +2. **Apply row filters** - Filter at scan time to reduce data transfer: + ```python + row_filter="key like 'IMOS/ANFOG/%'" + ``` + +3. **Use Polars** - The script returns Polars DataFrames for efficient in-memory operations + +## Output Format + +Results are saved to a timestamped text file. Example output: + +``` +================================================================================ +NetCDF Schema Variance Analysis +================================================================================ + +Timestamp: 2026-07-16T14:19:10.123456 +Query Prefix: IMOS/SOOP/SOOP-CO2% +Total Records: 731 +Total Distinct Schemas: 2 +Total Unique Variables: 22 + +================================================================================ +All Variables Found +================================================================================ + - AIRT_raw + - ATMP_uncorr_raw + - CO2_STD_Value + - ... + - xH2O_PPM_raw + +================================================================================ +Schema Variance Details +================================================================================ + +Schema #1 + Variables (20): ['AIRT_raw', 'ATMP_uncorr_raw', 'CO2_STD_Value', ...] + Record Count: 450 + Variables Unique to This Schema: + + Diff_Press_Equ_raw + + LabMain_sw_flow_raw + Variables Missing from This Schema: + - xCO2_ADJUSTED + Associated Prefixes (2): + - IMOS/SOOP/SOOP-CO2/VLMJ_Investigator/REALTIME/2022/11/ + - IMOS/SOOP/SOOP-CO2/VLMJ_Investigator/REALTIME/2023/01/ + +Schema #2 + Variables (22): ['AIRT_raw', 'ATMP_uncorr_raw', 'CO2_STD_Value', ...] + Record Count: 281 + Variables Unique to This Schema: + + xCO2_ADJUSTED + Variables Missing from This Schema: + Associated Prefixes (1): + - IMOS/SOOP/SOOP-CO2/VLMJ_Investigator/HISTORICAL/ +``` + +**Key features:** +- Schemas with identical variables in any order are now treated as the same schema +- "All Variables Found" section lists every variable across all schemas +- Each schema shows: + - Variable count and list + - Record count (number of files) + - Variables unique to this schema (marked with `+`) + - Variables missing from this schema that exist elsewhere (marked with `-`) + - Associated directory prefixes where this schema appears +- Helps identify which datasets have incomplete variable sets + +## Output Interpretation + +The `value_counts()` output shows: +- **Index**: Variable schema (as a list or string) +- **Count**: Number of records with that schema + +Example: +``` +variable_schema count +["temp", "sal", "depth"] 1250 +["temp", "sal", "depth", "cond"] 450 +["temp", "sal"] 200 +``` + +High variance (many unique schemas) indicates inconsistent data collection or processing. + +## Examples + +### Analyse a specific facility + +```python +df = query_with_filter( + table, + s3_prefix="IMOS/ANFOG/%", + selected_fields=("key", "facility", "variables") +) +print(df["variables"].value_counts(sort=True).head(10)) +``` + +Or from the command line: +```bash +uv run python analyse_netcdf_schema.py "IMOS/ANFOG/%" +``` + +### Find records with specific variables + +```python +df = query_with_filter(table, selected_fields=("key", "variables")) +df_with_temp = df.filter(pl.col("variables").str.contains("temp")) +print(f"Records with 'temp' variable: {len(df_with_temp)}") +``` + +### Group analysis by facility + +```python +df = query_with_filter(table, selected_fields=("key", "facility", "variables")) +grouped = df.group_by("facility").agg( + pl.col("variables").value_counts().alias("schemas") +) +print(grouped) +``` + +## Troubleshooting + +**Connection Issues:** +- Verify AWS credentials: `aws sts get-caller-identity` +- Check S3 Tables ARN is correct +- Ensure IAM role has S3 Tables access permissions + +**"Unable to locate AWS credentials" error:** +- Verify you've configured credentials using one of the methods above +- Check environment variables: `echo $AWS_ACCESS_KEY_ID` +- Try: `aws configure` to set up credentials + +**Empty Results:** +- Verify row filter syntax (uses SQL WHERE clause syntax) +- Check table name and namespace +- Ensure the S3 prefix contains data (try without prefix first) + +**Performance Issues:** +- Reduce selected fields +- Add stricter row filters +- Process in smaller batches if needed + +## Output Data Types + +The script returns Polars DataFrames where: +- `key` (str): S3 object key +- `size` (int): Object size in bytes +- `variables` (list): List of variable names in the NetCDF file +- `facility` (str): IMOS facility name +- `collection` (str): Legacy collection name + +## Related Resources + +- [Iceberg Documentation](https://iceberg.apache.org/) +- [Polars Documentation](https://docs.pola-rs.com/) +- [NetCDF Documentation](https://www.unidata.ucar.edu/software/netcdf/) diff --git a/src/data_index/cli/__init__.py b/src/data_index/cli/__init__.py new file mode 100644 index 0000000..f009e03 --- /dev/null +++ b/src/data_index/cli/__init__.py @@ -0,0 +1 @@ +"""CLI tools for data-index.""" diff --git a/src/data_index/cli/analyse_netcdf_schema.py b/src/data_index/cli/analyse_netcdf_schema.py new file mode 100644 index 0000000..34ef9f9 --- /dev/null +++ b/src/data_index/cli/analyse_netcdf_schema.py @@ -0,0 +1,334 @@ +#!/usr/bin/env python3 +""" +NetCDF Schema Variance Analysis CLI Tool + +Analyzes the structure and variance of NetCDF datasets stored in S3 Tables Iceberg catalog. +Identifies schema inconsistencies across collections and facilities. +""" + +from data_index.iceberg_config import S3TablesCatalogConfig, IcebergTableConfig +import polars as pl +import boto3 +import os +from datetime import datetime +from pathlib import Path +from typing import Optional +import typer +from rich.console import Console +from rich.table import Table + +app = typer.Typer(help="Analyze NetCDF schema variance in S3 Tables") +console = Console() + + +def load_table( + namespace: str = "data_index", + table_name: str = "structured_metadata_v5", + region: str = "ap-southeast-2", + arn: str = "arn:aws:s3tables:ap-southeast-2:704910415367:bucket/data-index", +): + """Load an Iceberg table from S3 Tables catalog.""" + # Setup AWS credentials from the current environment or IAM role + session = boto3.Session() + credentials = session.get_credentials() + + if credentials is None: + raise RuntimeError( + "Unable to locate AWS credentials. Please configure credentials via:\n" + " - AWS CLI: aws configure\n" + " - Environment variables: AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY\n" + " - IAM role (if running on EC2 or ECS)" + ) + + frozen = credentials.get_frozen_credentials() + + # Set environment variables for pyiceberg + os.environ["AWS_ACCESS_KEY_ID"] = frozen.access_key + os.environ["AWS_SECRET_ACCESS_KEY"] = frozen.secret_key + if frozen.token: + os.environ["AWS_SESSION_TOKEN"] = frozen.token + + catalog_config = S3TablesCatalogConfig(region=region, arn=arn) + table_config = IcebergTableConfig( + catalog_config=catalog_config, + namespace=namespace, + table_name=table_name, + ) + return table_config.load() + + +def query_with_filter( + table, + s3_prefix: str = None, + row_filter: str = None, + selected_fields: tuple = None, +): + """ + Query table with optional S3 prefix filter and column pruning. + + Args: + table: Iceberg table object + s3_prefix: S3 prefix to filter by (e.g., 'IMOS/ANFOG/%') + row_filter: Additional row filter predicate + selected_fields: Fields to retrieve (column pruning) + + Returns: + Polars DataFrame + """ + # Build combined row filter + filters = [] + if s3_prefix: + filters.append(f"key like '{s3_prefix}'") + if row_filter: + filters.append(f"({row_filter})") + + combined_filter = " AND ".join(filters) if filters else None + + scan = table.scan( + row_filter=combined_filter, + selected_fields=selected_fields, + ) + return scan.to_polars() + + +def analyze_schema_variance(df: pl.DataFrame, group_by: str = None): + """ + Analyze schema variance by examining variables column. + Schemas with the same variables (in any order) are considered equivalent. + + Args: + df: Polars DataFrame with variables column and key column + group_by: Optional column to group analysis by (e.g., 'facility', 'collection') + + Returns: + Dict with variance analysis results including prefix associations + """ + results = {} + + # Normalize schemas by sorting variables for comparison + # Create a new column with sorted variables as a string for grouping + def normalize_schema(variables): + if isinstance(variables, list): + return str(sorted(variables)) + elif isinstance(variables, str): + return str(sorted(eval(variables))) + else: + return str(sorted(variables)) + + df_normalized = df.with_columns( + pl.col("variables") + .map_elements(normalize_schema, return_dtype=pl.Utf8) + .alias("normalized_schema") + ) + + # Count records per normalized schema + schema_counts = df_normalized["normalized_schema"].value_counts(sort=True) + + # For each normalized schema, find the associated prefixes and collect variables + schema_prefixes = [] + all_variables = set() + + for schema_row in schema_counts.rows(named=True): + normalized_schema = schema_row["normalized_schema"] + count = schema_row["count"] + + # Find records with this normalized schema + records_with_schema = df_normalized.filter(pl.col("normalized_schema") == normalized_schema) + + # Extract directory path from keys by removing the filename + records_with_schema = records_with_schema.with_columns( + pl.col("key") + .str.replace(r'/[^/]*$', '/') + .alias("dir_prefix") + ) + + # Get unique prefixes for this schema, sorted + prefixes = records_with_schema["dir_prefix"].unique().sort().to_list() + + # Get actual variables from the original data (unsorted) + sample_vars = records_with_schema["variables"].to_list()[0] + if isinstance(sample_vars, str): + sample_vars = eval(sample_vars) + + # Collect all variables + all_variables.update(sample_vars) + + schema_prefixes.append({ + "schema": str(sorted(sample_vars)), + "schema_display": str(sample_vars), + "sorted_variables": sorted(sample_vars), + "count": count, + "prefixes": prefixes, + "num_prefixes": len(prefixes) + }) + + # Analyze which variables are common/unique to each schema + variable_analysis = [] + for idx, schema_info in enumerate(schema_prefixes): + schema_vars = set(schema_info["sorted_variables"]) + + # Find variables unique to this schema + other_vars = set().union(*[set(s["sorted_variables"]) for i, s in enumerate(schema_prefixes) if i != idx]) if len(schema_prefixes) > 1 else set() + unique_vars = schema_vars - other_vars + + # Find variables missing from this schema + missing_vars = all_variables - schema_vars + + schema_info["unique_variables"] = sorted(unique_vars) + schema_info["missing_variables"] = sorted(missing_vars) + + variable_analysis.append(schema_info) + + results["schema_variance"] = variable_analysis + results["total_schemas"] = len(variable_analysis) + results["total_records"] = len(df) + results["all_variables"] = sorted(all_variables) + + return results + + +def save_results(results: dict, s3_prefix: str = None): + """ + Save analysis results to a timestamped output file. + + Args: + results: Analysis results dictionary + s3_prefix: S3 prefix that was queried (used in filename) + + Returns: + Path to the output file + """ + # Generate timestamp and filename + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + prefix_part = s3_prefix.replace("/", "_").replace("%", "wildcard") if s3_prefix else "all" + output_file = f"netcdf_schema_analysis_{prefix_part}_{timestamp}.txt" + + # Write results + with open(output_file, "w") as f: + f.write("=" * 80 + "\n") + f.write("NetCDF Schema Variance Analysis\n") + f.write("=" * 80 + "\n\n") + + f.write(f"Timestamp: {datetime.now().isoformat()}\n") + f.write(f"Query Prefix: {s3_prefix or 'All data'}\n") + f.write(f"Total Records: {results['total_records']}\n") + f.write(f"Total Distinct Schemas: {results['total_schemas']}\n") + f.write(f"Total Unique Variables: {len(results['all_variables'])}\n\n") + + f.write("=" * 80 + "\n") + f.write("All Variables Found\n") + f.write("=" * 80 + "\n") + for var in results["all_variables"]: + f.write(f" - {var}\n") + f.write("\n") + + f.write("=" * 80 + "\n") + f.write("Schema Variance Details\n") + f.write("=" * 80 + "\n\n") + + for idx, schema_info in enumerate(results["schema_variance"], 1): + f.write(f"Schema #{idx}\n") + f.write(f" Variables ({len(schema_info['sorted_variables'])}): {schema_info['schema_display']}\n") + f.write(f" Record Count: {schema_info['count']}\n") + + if schema_info["unique_variables"]: + f.write(f" Variables Unique to This Schema:\n") + for var in schema_info["unique_variables"]: + f.write(f" + {var}\n") + + if schema_info["missing_variables"]: + f.write(f" Variables Missing from This Schema:\n") + for var in schema_info["missing_variables"]: + f.write(f" - {var}\n") + + f.write(f" Associated Prefixes ({schema_info['num_prefixes']}):\n") + for prefix in schema_info["prefixes"]: + f.write(f" - {prefix}\n") + f.write("\n") + + return output_file + + +@app.command() +def analyse( + s3_prefix: Optional[str] = typer.Argument( + None, + help="S3 prefix to analyse (e.g., 'IMOS/ANFOG/%', 'IMOS/SOOP/SOOP-CO2%')" + ), + output_dir: Optional[Path] = typer.Option( + None, + "--output-dir", + "-o", + help="Output directory for results (default: current directory)" + ) +): + """ + Analyze NetCDF schema variance in S3 Tables. + + Examines the structure of NetCDF datasets and identifies schema inconsistencies + across collections and facilities. + """ + try: + console.print("[bold blue]Loading S3 Table...[/bold blue]") + table = load_table() + + console.print("\n[bold]=== Querying Data ===[/bold]") + df_all = query_with_filter( + table, + s3_prefix=s3_prefix, + selected_fields=("key", "variables"), + ) + console.print(f"Total records: [green]{len(df_all)}[/green]") + if s3_prefix: + console.print(f"Prefix: [cyan]{s3_prefix}[/cyan]") + + console.print("\n[bold]=== Analyzing Schema Variance ===[/bold]") + results = analyze_schema_variance(df_all) + + console.print(f"Total distinct schemas: [yellow]{results['total_schemas']}[/yellow]") + console.print(f"Total unique variables: [yellow]{len(results['all_variables'])}[/yellow]") + + # Display top 10 schemas in a table + console.print(f"\n[bold]Top 10 schemas by record count:[/bold]") + table_display = Table(title="Schema Summary") + table_display.add_column("#", style="dim") + table_display.add_column("Records", justify="right") + table_display.add_column("Prefixes", justify="right") + table_display.add_column("Variables", style="cyan") + + for idx, schema_info in enumerate(results["schema_variance"][:10], 1): + table_display.add_row( + str(idx), + str(schema_info['count']), + str(schema_info['num_prefixes']), + schema_info['schema'][:60] + "..." if len(schema_info['schema']) > 60 else schema_info['schema'] + ) + console.print(table_display) + + console.print("\n[bold]=== Saving Results ===[/bold]") + # Change to output directory if specified + if output_dir: + output_dir.mkdir(parents=True, exist_ok=True) + original_cwd = os.getcwd() + os.chdir(output_dir) + try: + output_file = save_results(results, s3_prefix) + finally: + os.chdir(original_cwd) + output_path = output_dir / output_file + else: + output_file = save_results(results, s3_prefix) + output_path = Path(output_file) + + console.print(f"Results saved to: [green]{output_path.absolute()}[/green]") + + except RuntimeError as e: + console.print(f"[bold red]Error:[/bold red] {e}") + raise typer.Exit(code=1) + except Exception as e: + console.print(f"[bold red]Unexpected error:[/bold red] {e}") + raise typer.Exit(code=1) + + +if __name__ == "__main__": + app() From 845f245aa605888a31bbcfa08a263e3ccca2315f Mon Sep 17 00:00:00 2001 From: Michael Hemming Date: Thu, 16 Jul 2026 15:39:14 +1000 Subject: [PATCH 2/3] fix lint issues --- src/data_index/cli/analyse_netcdf_schema.py | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/src/data_index/cli/analyse_netcdf_schema.py b/src/data_index/cli/analyse_netcdf_schema.py index 34ef9f9..ae4d46f 100644 --- a/src/data_index/cli/analyse_netcdf_schema.py +++ b/src/data_index/cli/analyse_netcdf_schema.py @@ -6,17 +6,19 @@ Identifies schema inconsistencies across collections and facilities. """ -from data_index.iceberg_config import S3TablesCatalogConfig, IcebergTableConfig -import polars as pl -import boto3 import os from datetime import datetime from pathlib import Path from typing import Optional + +import boto3 +import polars as pl import typer from rich.console import Console from rich.table import Table +from data_index.iceberg_config import IcebergTableConfig, S3TablesCatalogConfig + app = typer.Typer(help="Analyze NetCDF schema variance in S3 Tables") console = Console() @@ -232,12 +234,12 @@ def save_results(results: dict, s3_prefix: str = None): f.write(f" Record Count: {schema_info['count']}\n") if schema_info["unique_variables"]: - f.write(f" Variables Unique to This Schema:\n") + f.write(" Variables Unique to This Schema:\n") for var in schema_info["unique_variables"]: f.write(f" + {var}\n") if schema_info["missing_variables"]: - f.write(f" Variables Missing from This Schema:\n") + f.write(" Variables Missing from This Schema:\n") for var in schema_info["missing_variables"]: f.write(f" - {var}\n") @@ -289,7 +291,7 @@ def analyse( console.print(f"Total unique variables: [yellow]{len(results['all_variables'])}[/yellow]") # Display top 10 schemas in a table - console.print(f"\n[bold]Top 10 schemas by record count:[/bold]") + console.print("[bold]Top 10 schemas by record count:[/bold]") table_display = Table(title="Schema Summary") table_display.add_column("#", style="dim") table_display.add_column("Records", justify="right") From b9b4f696eae8724999ecd79bc0cfafa60f8960b5 Mon Sep 17 00:00:00 2001 From: Michael Hemming Date: Thu, 16 Jul 2026 15:41:20 +1000 Subject: [PATCH 3/3] reformatted --- src/data_index/cli/analyse_netcdf_schema.py | 157 +++++++++++--------- 1 file changed, 89 insertions(+), 68 deletions(-) diff --git a/src/data_index/cli/analyse_netcdf_schema.py b/src/data_index/cli/analyse_netcdf_schema.py index ae4d46f..729cdf2 100644 --- a/src/data_index/cli/analyse_netcdf_schema.py +++ b/src/data_index/cli/analyse_netcdf_schema.py @@ -33,7 +33,7 @@ def load_table( # Setup AWS credentials from the current environment or IAM role session = boto3.Session() credentials = session.get_credentials() - + if credentials is None: raise RuntimeError( "Unable to locate AWS credentials. Please configure credentials via:\n" @@ -41,15 +41,15 @@ def load_table( " - Environment variables: AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY\n" " - IAM role (if running on EC2 or ECS)" ) - + frozen = credentials.get_frozen_credentials() - + # Set environment variables for pyiceberg os.environ["AWS_ACCESS_KEY_ID"] = frozen.access_key os.environ["AWS_SECRET_ACCESS_KEY"] = frozen.secret_key if frozen.token: os.environ["AWS_SESSION_TOKEN"] = frozen.token - + catalog_config = S3TablesCatalogConfig(region=region, arn=arn) table_config = IcebergTableConfig( catalog_config=catalog_config, @@ -67,13 +67,13 @@ def query_with_filter( ): """ Query table with optional S3 prefix filter and column pruning. - + Args: table: Iceberg table object s3_prefix: S3 prefix to filter by (e.g., 'IMOS/ANFOG/%') row_filter: Additional row filter predicate selected_fields: Fields to retrieve (column pruning) - + Returns: Polars DataFrame """ @@ -83,9 +83,9 @@ def query_with_filter( filters.append(f"key like '{s3_prefix}'") if row_filter: filters.append(f"({row_filter})") - + combined_filter = " AND ".join(filters) if filters else None - + scan = table.scan( row_filter=combined_filter, selected_fields=selected_fields, @@ -116,71 +116,83 @@ def normalize_schema(variables): return str(sorted(eval(variables))) else: return str(sorted(variables)) - + df_normalized = df.with_columns( pl.col("variables") .map_elements(normalize_schema, return_dtype=pl.Utf8) .alias("normalized_schema") ) - + # Count records per normalized schema schema_counts = df_normalized["normalized_schema"].value_counts(sort=True) - + # For each normalized schema, find the associated prefixes and collect variables schema_prefixes = [] all_variables = set() - + for schema_row in schema_counts.rows(named=True): normalized_schema = schema_row["normalized_schema"] count = schema_row["count"] - + # Find records with this normalized schema - records_with_schema = df_normalized.filter(pl.col("normalized_schema") == normalized_schema) - + records_with_schema = df_normalized.filter( + pl.col("normalized_schema") == normalized_schema + ) + # Extract directory path from keys by removing the filename records_with_schema = records_with_schema.with_columns( - pl.col("key") - .str.replace(r'/[^/]*$', '/') - .alias("dir_prefix") + pl.col("key").str.replace(r"/[^/]*$", "/").alias("dir_prefix") ) - + # Get unique prefixes for this schema, sorted prefixes = records_with_schema["dir_prefix"].unique().sort().to_list() - + # Get actual variables from the original data (unsorted) sample_vars = records_with_schema["variables"].to_list()[0] if isinstance(sample_vars, str): sample_vars = eval(sample_vars) - + # Collect all variables all_variables.update(sample_vars) - - schema_prefixes.append({ - "schema": str(sorted(sample_vars)), - "schema_display": str(sample_vars), - "sorted_variables": sorted(sample_vars), - "count": count, - "prefixes": prefixes, - "num_prefixes": len(prefixes) - }) - + + schema_prefixes.append( + { + "schema": str(sorted(sample_vars)), + "schema_display": str(sample_vars), + "sorted_variables": sorted(sample_vars), + "count": count, + "prefixes": prefixes, + "num_prefixes": len(prefixes), + } + ) + # Analyze which variables are common/unique to each schema variable_analysis = [] for idx, schema_info in enumerate(schema_prefixes): schema_vars = set(schema_info["sorted_variables"]) - + # Find variables unique to this schema - other_vars = set().union(*[set(s["sorted_variables"]) for i, s in enumerate(schema_prefixes) if i != idx]) if len(schema_prefixes) > 1 else set() + other_vars = ( + set().union( + *[ + set(s["sorted_variables"]) + for i, s in enumerate(schema_prefixes) + if i != idx + ] + ) + if len(schema_prefixes) > 1 + else set() + ) unique_vars = schema_vars - other_vars - + # Find variables missing from this schema missing_vars = all_variables - schema_vars - + schema_info["unique_variables"] = sorted(unique_vars) schema_info["missing_variables"] = sorted(missing_vars) - + variable_analysis.append(schema_info) - + results["schema_variance"] = variable_analysis results["total_schemas"] = len(variable_analysis) results["total_records"] = len(df) @@ -192,88 +204,91 @@ def normalize_schema(variables): def save_results(results: dict, s3_prefix: str = None): """ Save analysis results to a timestamped output file. - + Args: results: Analysis results dictionary s3_prefix: S3 prefix that was queried (used in filename) - + Returns: Path to the output file """ # Generate timestamp and filename timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") - prefix_part = s3_prefix.replace("/", "_").replace("%", "wildcard") if s3_prefix else "all" + prefix_part = ( + s3_prefix.replace("/", "_").replace("%", "wildcard") if s3_prefix else "all" + ) output_file = f"netcdf_schema_analysis_{prefix_part}_{timestamp}.txt" - + # Write results with open(output_file, "w") as f: f.write("=" * 80 + "\n") f.write("NetCDF Schema Variance Analysis\n") f.write("=" * 80 + "\n\n") - + f.write(f"Timestamp: {datetime.now().isoformat()}\n") f.write(f"Query Prefix: {s3_prefix or 'All data'}\n") f.write(f"Total Records: {results['total_records']}\n") f.write(f"Total Distinct Schemas: {results['total_schemas']}\n") f.write(f"Total Unique Variables: {len(results['all_variables'])}\n\n") - + f.write("=" * 80 + "\n") f.write("All Variables Found\n") f.write("=" * 80 + "\n") for var in results["all_variables"]: f.write(f" - {var}\n") f.write("\n") - + f.write("=" * 80 + "\n") f.write("Schema Variance Details\n") f.write("=" * 80 + "\n\n") - + for idx, schema_info in enumerate(results["schema_variance"], 1): f.write(f"Schema #{idx}\n") - f.write(f" Variables ({len(schema_info['sorted_variables'])}): {schema_info['schema_display']}\n") + f.write( + f" Variables ({len(schema_info['sorted_variables'])}): {schema_info['schema_display']}\n" + ) f.write(f" Record Count: {schema_info['count']}\n") - + if schema_info["unique_variables"]: f.write(" Variables Unique to This Schema:\n") for var in schema_info["unique_variables"]: f.write(f" + {var}\n") - + if schema_info["missing_variables"]: f.write(" Variables Missing from This Schema:\n") for var in schema_info["missing_variables"]: f.write(f" - {var}\n") - + f.write(f" Associated Prefixes ({schema_info['num_prefixes']}):\n") for prefix in schema_info["prefixes"]: f.write(f" - {prefix}\n") f.write("\n") - + return output_file @app.command() def analyse( s3_prefix: Optional[str] = typer.Argument( - None, - help="S3 prefix to analyse (e.g., 'IMOS/ANFOG/%', 'IMOS/SOOP/SOOP-CO2%')" + None, help="S3 prefix to analyse (e.g., 'IMOS/ANFOG/%', 'IMOS/SOOP/SOOP-CO2%')" ), output_dir: Optional[Path] = typer.Option( None, "--output-dir", "-o", - help="Output directory for results (default: current directory)" - ) + help="Output directory for results (default: current directory)", + ), ): """ Analyze NetCDF schema variance in S3 Tables. - + Examines the structure of NetCDF datasets and identifies schema inconsistencies across collections and facilities. """ try: console.print("[bold blue]Loading S3 Table...[/bold blue]") table = load_table() - + console.print("\n[bold]=== Querying Data ===[/bold]") df_all = query_with_filter( table, @@ -283,13 +298,17 @@ def analyse( console.print(f"Total records: [green]{len(df_all)}[/green]") if s3_prefix: console.print(f"Prefix: [cyan]{s3_prefix}[/cyan]") - + console.print("\n[bold]=== Analyzing Schema Variance ===[/bold]") results = analyze_schema_variance(df_all) - - console.print(f"Total distinct schemas: [yellow]{results['total_schemas']}[/yellow]") - console.print(f"Total unique variables: [yellow]{len(results['all_variables'])}[/yellow]") - + + console.print( + f"Total distinct schemas: [yellow]{results['total_schemas']}[/yellow]" + ) + console.print( + f"Total unique variables: [yellow]{len(results['all_variables'])}[/yellow]" + ) + # Display top 10 schemas in a table console.print("[bold]Top 10 schemas by record count:[/bold]") table_display = Table(title="Schema Summary") @@ -297,16 +316,18 @@ def analyse( table_display.add_column("Records", justify="right") table_display.add_column("Prefixes", justify="right") table_display.add_column("Variables", style="cyan") - + for idx, schema_info in enumerate(results["schema_variance"][:10], 1): table_display.add_row( str(idx), - str(schema_info['count']), - str(schema_info['num_prefixes']), - schema_info['schema'][:60] + "..." if len(schema_info['schema']) > 60 else schema_info['schema'] + str(schema_info["count"]), + str(schema_info["num_prefixes"]), + schema_info["schema"][:60] + "..." + if len(schema_info["schema"]) > 60 + else schema_info["schema"], ) console.print(table_display) - + console.print("\n[bold]=== Saving Results ===[/bold]") # Change to output directory if specified if output_dir: @@ -321,9 +342,9 @@ def analyse( else: output_file = save_results(results, s3_prefix) output_path = Path(output_file) - + console.print(f"Results saved to: [green]{output_path.absolute()}[/green]") - + except RuntimeError as e: console.print(f"[bold red]Error:[/bold red] {e}") raise typer.Exit(code=1)