""" Metrics collection and logging for the owned clip analysis backfill job. This module centralizes all logging, metrics publishing, and progress tracking functionality that was previously scattered throughout the monolithic script. """ import json import time from datetime import datetime from typing import Dict, Any, Optional import boto3 from pyspark.context import SparkContext from config.job_config import JobConfig, CloudWatchConfig class MetricsCollector: """ Centralized metrics collection and logging for the backfill job. This class handles: - Structured JSON logging with context - CloudWatch metrics publishing - Progress tracking with Spark accumulators - Performance metrics and timing - Checkpoint logging for recovery """ def __init__(self, config: JobConfig, spark_context: SparkContext = None): """ Initialize the metrics collector. Args: config: Job configuration spark_context: Spark context for distributed accumulators (None on worker nodes) """ self.config = config self.job_name = config.job_name self.environment = config.environment # Initialize CloudWatch client self.cloudwatch = boto3.client('cloudwatch', region_name=config.cloudwatch.region) # Initialize Spark accumulators for distributed stats tracking (only on driver) if spark_context is not None: self.processed_rows_acc = spark_context.accumulator(0) self.successful_inserts_acc = spark_context.accumulator(0) self.skipped_records_acc = spark_context.accumulator(0) self.batches_processed_acc = spark_context.accumulator(0) else: # On worker nodes, use None (metrics will be local only) self.processed_rows_acc = None self.successful_inserts_acc = None self.skipped_records_acc = None self.batches_processed_acc = None # Local state for progress tracking self._global_stats = { "total_rows": 0, "start_time": None, "last_progress_log": None, } def set_total_rows(self, total_rows: int): """Set the total number of rows for progress calculation.""" self._global_stats["total_rows"] = total_rows def set_start_time(self, start_time: datetime): """Set the job start time for progress calculation.""" self._global_stats["start_time"] = start_time def log_structured(self, level: str, message: str, **kwargs): """ Log structured JSON messages for better parsing and monitoring. Args: level: Log level (INFO, ERROR, WARNING) message: Log message **kwargs: Additional structured data """ timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3] log_entry = { "timestamp": timestamp, "level": level.upper(), "message": message, "job": self.job_name, "environment": self.environment, **kwargs } print(json.dumps(log_entry)) def log_info(self, message: str, **kwargs): """Log info level message with structured data.""" self.log_structured("INFO", message, **kwargs) def log_error(self, message: str, **kwargs): """Log error level message with structured data.""" self.log_structured("ERROR", message, **kwargs) def log_warning(self, message: str, **kwargs): """Log warning level message with structured data.""" self.log_structured("WARNING", message, **kwargs) def update_stats(self, stat_name: str, increment: int = 1, **additional_stats): """ Update distributed statistics using Spark accumulators. Args: stat_name: Name of the statistic to update increment: Amount to increment (0 for no increment) **additional_stats: Additional local stats to update """ # Update accumulator-based stats (thread-safe distributed, only on driver) if increment != 0: if stat_name == "processed_rows" and self.processed_rows_acc is not None: self.processed_rows_acc.add(increment) elif stat_name == "successful_inserts" and self.successful_inserts_acc is not None: self.successful_inserts_acc.add(increment) elif stat_name == "skipped_records" and self.skipped_records_acc is not None: self.skipped_records_acc.add(increment) elif stat_name == "batches_processed" and self.batches_processed_acc is not None: self.batches_processed_acc.add(increment) # Update non-distributed stats (driver-only) for key, value in additional_stats.items(): if key in self._global_stats: self._global_stats[key] = value def get_progress_info(self) -> Dict[str, Any]: """ Get current progress information with calculations using accumulators. Returns: Dictionary containing progress statistics """ # Get values from accumulators (safe to call from driver) stats = { "total_rows": self._global_stats["total_rows"], "processed_rows": self.processed_rows_acc.value if self.processed_rows_acc else 0, "successful_inserts": self.successful_inserts_acc.value if self.successful_inserts_acc else 0, "skipped_records": self.skipped_records_acc.value if self.skipped_records_acc else 0, "batches_processed": self.batches_processed_acc.value if self.batches_processed_acc else 0, "start_time": self._global_stats["start_time"], "last_progress_log": self._global_stats["last_progress_log"] } # Calculate progress percentage if stats["total_rows"] > 0: progress_pct = (stats["processed_rows"] / stats["total_rows"]) * 100 else: progress_pct = 0 # Calculate timing and throughput elapsed_time = None throughput_per_sec = 0 eta_seconds = None if stats["start_time"]: elapsed_time = (datetime.now() - stats["start_time"]).total_seconds() if elapsed_time > 0: throughput_per_sec = stats["processed_rows"] / elapsed_time if throughput_per_sec > 0 and stats["total_rows"] > stats["processed_rows"]: remaining_rows = stats["total_rows"] - stats["processed_rows"] eta_seconds = remaining_rows / throughput_per_sec return { **stats, "progress_percentage": round(progress_pct, 2), "elapsed_seconds": elapsed_time, "throughput_rows_per_sec": round(throughput_per_sec, 2) if throughput_per_sec else 0, "eta_seconds": round(eta_seconds) if eta_seconds else None, "eta_human": self._format_duration(eta_seconds) if eta_seconds else None } def _format_duration(self, seconds: float) -> str: """ Format duration in human readable format. Args: seconds: Duration in seconds Returns: Human-readable duration string """ if seconds < 60: return f"{int(seconds)}s" elif seconds < 3600: minutes = int(seconds // 60) remaining_seconds = int(seconds % 60) return f"{minutes}m {remaining_seconds}s" else: hours = int(seconds // 3600) minutes = int((seconds % 3600) // 60) return f"{hours}h {minutes}m" def log_batch_performance(self, batch_size: int, processing_time_seconds: float, insert_time_seconds: Optional[float] = None, **additional_metrics): """ Log detailed batch performance metrics. Args: batch_size: Number of records in the batch processing_time_seconds: Total processing time insert_time_seconds: Database insert time **additional_metrics: Additional metrics to log """ throughput = batch_size / processing_time_seconds if processing_time_seconds > 0 else 0 metrics = { "batch_size": batch_size, "processing_time_seconds": round(processing_time_seconds, 3), "rows_per_second": round(throughput, 2), **additional_metrics } if insert_time_seconds: metrics["insert_time_seconds"] = round(insert_time_seconds, 3) if insert_time_seconds > 0: metrics["insert_rows_per_second"] = round(batch_size / insert_time_seconds, 2) self.log_info("Batch Performance Metrics", **metrics) def publish_cloudwatch_metrics(self, metric_data: Optional[Dict[str, Any]] = None): """ Publish custom metrics to CloudWatch for monitoring. Args: metric_data: Optional specific metrics to publish. If None, uses current progress. """ try: if metric_data is None: metric_data = self.get_progress_info() metrics = [] namespace = self.config.cloudwatch.namespace job_dimension = {"Name": "JobName", "Value": self.job_name} env_dimension = {"Name": "Environment", "Value": self.environment} # Add progress percentage metric if "progress_percentage" in metric_data: metrics.append({ "MetricName": "ProgressPercentage", "Value": metric_data["progress_percentage"], "Unit": "Percent", "Dimensions": [job_dimension, env_dimension] }) # Add throughput metric if "throughput_rows_per_sec" in metric_data: metrics.append({ "MetricName": "ThroughputRowsPerSecond", "Value": metric_data["throughput_rows_per_sec"], "Unit": "Count/Second", "Dimensions": [job_dimension, env_dimension] }) # Add processed rows metric if "processed_rows" in metric_data: metrics.append({ "MetricName": "ProcessedRows", "Value": metric_data["processed_rows"], "Unit": "Count", "Dimensions": [job_dimension, env_dimension] }) # Add successful inserts metric if "successful_inserts" in metric_data: metrics.append({ "MetricName": "SuccessfulInserts", "Value": metric_data["successful_inserts"], "Unit": "Count", "Dimensions": [job_dimension, env_dimension] }) # Add skipped records metric if "skipped_records" in metric_data: metrics.append({ "MetricName": "SkippedRecords", "Value": metric_data["skipped_records"], "Unit": "Count", "Dimensions": [job_dimension, env_dimension] }) # Publish metrics in batches (CloudWatch limit is 20 metrics per call) batch_size = self.config.cloudwatch.metrics_batch_size for i in range(0, len(metrics), batch_size): batch = metrics[i:i+batch_size] self.cloudwatch.put_metric_data( Namespace=namespace, MetricData=batch ) self.log_info("CloudWatch Metrics Published", metrics_count=len(metrics), namespace=namespace ) except Exception as e: self.log_warning("Failed to publish CloudWatch metrics", error=str(e)[:200] ) def log_checkpoint(self, checkpoint_name: str, **additional_data): """ Log significant checkpoints for recovery and monitoring. Args: checkpoint_name: Name of the checkpoint **additional_data: Additional checkpoint-specific data """ progress = self.get_progress_info() checkpoint_data = { "checkpoint": checkpoint_name, "timestamp": datetime.now().isoformat(), "progress_percentage": progress["progress_percentage"], "processed_rows": progress["processed_rows"], "total_rows": progress["total_rows"], "successful_inserts": progress["successful_inserts"], "skipped_records": progress["skipped_records"], "elapsed_seconds": progress["elapsed_seconds"], **additional_data } self.log_info("CHECKPOINT", **checkpoint_data) # Also publish checkpoint metrics to CloudWatch self.publish_cloudwatch_metrics(progress) def log_date_completed(self, date_str: str, date_stats: Dict[str, Any]): """ Log completion of a specific date's processing for recovery tracking. Args: date_str: Date that was completed (YYYY-MM-DD) date_stats: Statistics specific to this date's processing """ progress = self.get_progress_info() date_completion_data = { "checkpoint": "DATE_COMPLETED", "completed_date": date_str, "timestamp": datetime.now().isoformat(), "date_processing_time_seconds": date_stats.get("processing_time_seconds", 0), "date_rows_processed": date_stats.get("rows_processed", 0), "date_successful_inserts": date_stats.get("successful_inserts", 0), "date_partitions_processed": date_stats.get("partitions_processed", 0), # Overall job progress "overall_progress_percentage": progress["progress_percentage"], "overall_processed_rows": progress["processed_rows"], "overall_successful_inserts": progress["successful_inserts"], "job_elapsed_seconds": progress["elapsed_seconds"] } self.log_info("DATE_PROCESSING_COMPLETED", **date_completion_data) def log_batch_completion(self, partition_id: str, batch_stats: Dict[str, Any]): """ Log batch processing completion with comprehensive statistics including throttling impact. Args: partition_id: ID of the partition being processed batch_stats: Dictionary containing batch processing statistics """ self.log_info("Batch Processing Complete", partition_id=partition_id, batch_size=batch_stats.get("batch_size", 0), successful_inserts=batch_stats.get("successful_inserts", 0), duplicate_records=batch_stats.get("duplicate_records", 0), processing_path=batch_stats.get("processing_path", "unknown"), raw_processing_time_ms=batch_stats.get("raw_processing_time_ms", 0), total_batch_time_ms=batch_stats.get("total_batch_time_ms", 0), throttle_delay_ms=batch_stats.get("throttle_delay_ms", 0), insert_time_ms=batch_stats.get("insert_time_ms", 0), raw_throughput_per_sec=batch_stats.get("raw_throughput_per_sec", 0), effective_throughput_per_sec=batch_stats.get("effective_throughput_per_sec", 0) ) def log_partition_completion(self, partition_id: str, partition_stats: Dict[str, Any]): """ Log partition processing completion with comprehensive statistics. Args: partition_id: ID of the partition that completed partition_stats: Dictionary containing partition statistics """ duration = partition_stats.get("duration_seconds", 0) processed_rows = partition_stats.get("processed_rows", 0) successful_inserts = partition_stats.get("successful_inserts", 0) # Calculate success rate success_rate = "0%" if processed_rows > 0: success_rate = f"{(successful_inserts / processed_rows * 100):.1f}%" # Format duration duration_human = self._format_duration(duration) if duration >= 60 else f"{duration:.1f}s" self.log_info("Partition Processing Complete", partition_id=partition_id, processed_rows=processed_rows, successful_inserts=successful_inserts, skipped_records=partition_stats.get("skipped_records", 0), success_rate=success_rate, duration_seconds=round(duration, 2), duration_human=duration_human, rows_per_second=round(processed_rows / duration, 1) if duration > 0 else 0, inserts_per_second=round(successful_inserts / duration, 1) if duration > 0 else 0, connection_refreshes=partition_stats.get("connection_refreshes", 0) )