Cash Collection Wiki

Space shortcuts

Page tree

GitLab: GitLab


ALMINA Plant Data Pipeline – Technical Documentation

Overview

This pipeline automates the extraction, transformation, validation, and loading (ETL) of daily plant data from ALMINA’s Excel workbooks into a BigQuery summary table. It is designed for both batch and event-driven (GCP Cloud Function) operation, with robust error handling, logging, and schema management.

Architecture & Workflow

High-level steps:

  1. Ingestion: Excel files are read from Google Cloud Storage (GCS), either in batch or via Cloud Function trigger.
  2. Mapping: The MappingTest worksheet defines how to extract and map data from daily sheets.
  3. Extraction & Transformation: Data is extracted per shift, cleaned, and converted to the correct types.
  4. Validation: Data quality is checked, and metadata is attached.
  5. Loading: The processed DataFrame is uploaded to BigQuery, with schema alignment and idempotency.
  6. Logging: All steps are logged with structured, context-rich messages.

Configuration

  • All configuration is centralized in the Settings dataclass, loaded from environment variables.
  • Required: GCP_PROJECT_ID, GCS_BUCKET_NAME, BQ_DATASET_ID
  • Optional: BQ_TABLE_ID, CUSTOMER_ID, GOOGLE_APPLICATION_CREDENTIALS
  • Retry, partitioning, and clustering options are configurable.

Module Reference

Main Script (main.py)

  • Purpose: Entry point for both local batch and GCP Cloud Function execution.
  • Key Functions:
    • main(): Batch processing from local Excel file to BigQuery.
    • almina_extra_data_mapping_trigger(cloud_event): Cloud Function for GCS-triggered processing.
    • prepare_dataframe_for_bigquery(df): Ensures DataFrame is BigQuery-compatible (types, nulls, etc.).
  • Dependencies: All core modules below.

SummaryDataProcessor

  • Purpose: Orchestrates the extraction and transformation of plant data from Excel workbooks using mapping rules.
  • Key Methods:
    • process_workbook(raw_workbook_bytes, mapping_csv_rows, file_name, upload_time_to_bucket): Main entry point; processes a workbook and returns a DataFrame.
    • _identify_daily_sheets(workbook): Finds daily sheets by name pattern.
    • _process_sheet(...): Extracts data for both shifts from a sheet.
    • _calculate_derived_fields(record, sheet_name, shift): Derives date, time, shift, and other fields.
    • _extract_cell_value(...): Extracts and cleans a value from a cell or range.
    • _add_metadata(df, ...): Adds metadata columns.
    • _calculate_quality_score(df, error_count): Computes a data quality score.
  • Dependencies: Mapping rules, data transformer, logging, sanitizers.

Field Mapping & Mapping Rules

  • Purpose: Defines how to interpret the MappingTest worksheet and map fields to cell references.
  • Key Classes:
    • CellReference: Represents a cell or range, with optional sheet override.
    • FieldMappingRule: Holds mapping info for a single field (cell refs, clues, required, examples).
    • MappingConfig: Aggregates all field mapping rules.
    • MappingTestParser: Parses mapping rows into a MappingConfig.
  • Key Function:
    • parse_mapping_test_rows(rows): Parses mapping rows into config.

Data Transformation

  • Purpose: Converts raw extracted values to the correct types for BigQuery.
  • Key Class:
    • DataTransformer: Handles numeric, date, timestamp, and string parsing, with error handling and logging.
  • Key Function:
    • transform_dataframe(df, schema_types, logger): Applies type transformation to an entire DataFrame.

BigQuery Schema & Utilities

  • Purpose: Defines and generates BigQuery schemas based on mapping fields.
  • Key Classes/Functions:
    • SummaryTableSchema: Canonical schema with field/type lists and utilities.
    • build_schema_from_mapping_fields(field_names): Builds schema from mapping fields, inferring types.
    • _infer_field_type(clean_name, original_name): Infers BigQuery type from field name.
    • SummaryBigQuerySchema: Generates schema from mapping config, using example values for type inference.

BigQuery Service

  • Purpose: Handles all interactions with BigQuery: table creation, schema management, and data upload.
  • Key Class:
    • SummaryBigQueryService
      • ensure_table_exists(field_names, ...): Ensures the summary table exists and is up-to-date.
      • upload_dataframe(df, ...): Uploads DataFrame, with idempotency and schema alignment.
      • _upsert_via_merge(df): Upserts using staging table and MERGE.
      • _prepare_dataframe_for_upload(df): Cleans and converts DataFrame for upload.
      • _final_pyarrow_safety_check(df): Ensures PyArrow compatibility.
      • _align_dataframe_with_bq_schema(df): Aligns DataFrame types with BQ schema.
      • _truncate_datetime_columns_to_microseconds(df): Ensures timestamp precision.
      • _ensure_record_id(df): Adds idempotency key.
      • query_recent_data(limit, days_back): Queries recent data for validation/testing.
      • get_table_info(): Returns table metadata.

GCP Storage Service

  • Purpose: Encapsulates all Google Cloud Storage operations.
  • Key Class:
    • GCPStorageService
      • list_xlsx_files(): Lists all .xlsx files in the bucket.
      • download_file(file_name): Downloads a file and returns its bytes and upload time.
  • Implements: Retry logic for robustness.

Logging Utilities

  • Purpose: Provides structured logging for both local and cloud environments.
  • Key Class:
    • StructuredLogger: Logs messages with severity, component, timestamp, and extra metadata.
  • Key Function:
    • setup_logging(component): Configures and returns a StructuredLogger.

Settings & Configuration

  • Purpose: Centralizes all configuration and environment variables.
  • Key Class:
    • Settings: Immutable dataclass for all settings.
      • from_env(): Loads from environment.
      • validate(): Ensures required settings are present.
  • Key Functions:
    • get_settings(): Returns singleton settings instance.
  • Global: settings is the default instance.

Sanitizers

  • Purpose: Ensures all column names are BigQuery-safe and snake_case.
  • Key Function:
    • sanitize_column_name(original_name): Converts any string to a valid BigQuery column name.

Retry Utilities

  • Purpose: Provides a decorator for retrying functions with exponential backoff.
  • Key Function:
    • retry_with_backoff(max_retries, backoff_factor): Decorator for robust retry logic, with logging.

Data Flow Summary

  1. File Ingestion:
    • Files are listed/downloaded from GCS via GCPStorageService.
  2. Mapping Extraction:
    • The MappingTest worksheet is parsed into a MappingConfig using MappingTestParser.
  3. Data Extraction:
    • For each daily sheet and shift, values are extracted using mapping rules.
  4. Transformation:
    • Raw values are converted to correct types using DataTransformer and transform_dataframe.
  5. Validation & Metadata:
    • Data quality is checked, and metadata columns are added.
  6. BigQuery Upload:
    • DataFrame is cleaned, aligned, and uploaded to BigQuery using SummaryBigQueryService.
  7. Logging:
    • All steps are logged with context for traceability.

Error Handling & Logging

  • StructuredLogger ensures all logs are context-rich and compatible with GCP logging.
  • retry_with_backoff ensures transient errors are retried with exponential backoff.
  • Validation errors are accumulated and attached as metadata to each record.
  • Critical failures are logged and raised, preventing silent data loss.

Extensibility & Best Practices

  • Mapping-driven: New fields can be added by updating the MappingTest worksheet and field mappings.
  • Schema evolution: The pipeline can add new columns to BigQuery as needed.
  • Separation of concerns: Each module handles a single responsibility.
  • Cloud-native: Designed for both local and GCP Cloud Function execution.

Security & Compliance

  • Authentication: Uses service accounts and environment variables for secure GCP access.
  • Data Handling: All sensitive operations are logged; errors are surfaced for review.
  • Compliance: Ensure GCP IAM roles and data residency requirements are met for your organization.

Appendix: Example Usage

Local Batch Processing

export GCP_PROJECT_ID=your-project
export GCS_BUCKET_NAME=your-bucket
export BQ_DATASET_ID=your-dataset
python main.py

Cloud Function Deployment

  • Deploy almina_extra_data_mapping_trigger as a GCP Cloud Function (Gen2) with a GCS trigger.
  • Ensure all environment variables are set in the Cloud Function configuration.