""" Main orchestrator for the owned clip analysis backfill job. This module coordinates all components to execute the backfill operation with proper error handling, progress tracking, and resource management. """ import uuid import time from datetime import datetime from typing import List, Dict, Any, Iterator from pyspark.sql import DataFrame from config.job_config import JobConfig from core.database_manager import DatabaseManager from core.batch_processor import BatchProcessor from core.s3_data_loader import S3DataLoader from core.partition_processor import create_partition_processor from lib.metrics_collector import MetricsCollector from lib.retry_handler import RetryHandler from lib.date_helpers import generate_date_range, get_date_range_info from lib.exceptions import BackfillJobError, S3Error, DatabaseError class BackfillJob: """ Main orchestrator for the owned clip analysis backfill job. This class coordinates all components to: - Load and validate configuration - Discover and load S3 data - Process data in parallel partitions - Track progress and handle errors - Provide comprehensive logging and metrics """ def __init__(self, config: JobConfig, glue_context, spark_context, metrics_collector: MetricsCollector): """ Initialize the backfill job with all required components. Args: config: Job configuration glue_context: AWS Glue context spark_context: Spark context for distributed processing metrics_collector: Metrics collector for logging and monitoring """ self.config = config self.glue_context = glue_context self.spark_context = spark_context self.metrics_collector = metrics_collector # Initialize components using injected metrics collector self.database_manager = DatabaseManager(config, self.metrics_collector) self.retry_handler = RetryHandler(config, self.metrics_collector) self.batch_processor = BatchProcessor( config, self.database_manager, self.metrics_collector, self.retry_handler ) self.s3_data_loader = S3DataLoader(config, glue_context, self.metrics_collector) # Job state self.job_start_time = datetime.now() self.total_rows = 0 def run(self) -> Dict[str, Any]: """ Execute the complete backfill job using day-by-day processing. Returns: Dictionary containing job execution results and statistics Raises: BackfillJobError: If job execution fails """ try: # Initialize job self._initialize_job() # Process each day in the date range individually self._process_date_range_daily() # Complete job and return results return self._complete_job_successfully() except Exception as e: return self._handle_job_failure(e) finally: self._cleanup_resources() def _initialize_job(self): """Initialize job state and logging.""" self.metrics_collector.set_start_time(self.job_start_time) # Get and store date range information for reuse throughout job self.date_range_info = get_date_range_info( self.config.backfill_start_date, self.config.backfill_end_date, self.config.s3.partition_date_format ) self.metrics_collector.log_checkpoint("JOB_STARTED", table_name=self.config.table_name, backfill_start_date=self.config.backfill_start_date, backfill_end_date=self.config.backfill_end_date, environment=self.config.environment, account_id=self.config.account_id, s3_bucket=self.config.s3_bucket, parameter_info="Day-by-day processing with completion tracking", **self.date_range_info ) def _process_date_range_daily(self): """ Process the backfill date range day-by-day with completion tracking. This method replaces the bulk processing approach with individual date processing, allowing for granular progress tracking and recovery. """ date_count = 0 dates_processed = 0 dates_skipped = 0 # Generate and process each date in the range (newest to oldest) for current_date in generate_date_range( self.config.backfill_start_date, self.config.backfill_end_date, self.config.s3.partition_date_format, reverse=True ): date_count += 1 date_start_time = time.time() self.metrics_collector.log_info("Starting date processing", current_date=current_date, date_progress=f"{date_count} of {self.date_range_info['total_days']}" ) try: # Process single date date_stats = self._process_single_date(current_date) if date_stats["rows_processed"] > 0: dates_processed += 1 # Log date completion for recovery tracking date_stats["processing_time_seconds"] = time.time() - date_start_time self.metrics_collector.log_date_completed(current_date, date_stats) self.metrics_collector.log_info("Date processing completed", current_date=current_date, rows_processed=date_stats["rows_processed"], successful_inserts=date_stats["successful_inserts"], processing_time_seconds=date_stats["processing_time_seconds"] ) else: dates_skipped += 1 self.metrics_collector.log_info("Date skipped - no data found", current_date=current_date ) except Exception as e: self.metrics_collector.log_error("Date processing failed", current_date=current_date, error_type=type(e).__name__, error_message=str(e)[:300] ) raise BackfillJobError(f"Failed to process date {current_date}: {e}") from e # Log overall date range completion self.metrics_collector.log_checkpoint("DATE_RANGE_PROCESSING_COMPLETE", total_dates_in_range=date_count, dates_processed=dates_processed, dates_skipped=dates_skipped ) def _process_single_date(self, target_date: str) -> Dict[str, Any]: """ Process data for a single date. Args: target_date: Date to process in YYYY-MM-DD format Returns: Dictionary containing processing statistics for this date """ # Discover partitions for this specific date partition_paths = self.s3_data_loader.discover_partitions_for_date(target_date) if not partition_paths: return { "rows_processed": 0, "successful_inserts": 0, "partitions_processed": 0 } # Load and prepare data for this date only dataframe = self._load_and_prepare_data(partition_paths) if dataframe is None: return { "rows_processed": 0, "successful_inserts": 0, "partitions_processed": len(partition_paths) } # Get initial statistics date_rows = dataframe.count() initial_progress = self.metrics_collector.get_progress_info() # Process this date's data self._process_data_parallel(dataframe) # Calculate statistics for this date final_progress = self.metrics_collector.get_progress_info() return { "rows_processed": date_rows, "successful_inserts": final_progress["successful_inserts"] - initial_progress["successful_inserts"], "partitions_processed": len(partition_paths) } def _discover_s3_data(self) -> List[str]: """Discover S3 partitions for the configured date range.""" self.metrics_collector.log_info("Discovering S3 partitions", s3_bucket=self.config.s3_bucket, backfill_start_date=self.config.backfill_start_date, backfill_end_date=self.config.backfill_end_date ) partition_paths = self.s3_data_loader.discover_partitions( self.config.backfill_start_date, self.config.backfill_end_date ) if partition_paths: partition_summary = self.s3_data_loader.get_partition_summary(partition_paths) self.metrics_collector.log_info("S3 partition discovery completed", **partition_summary) else: self.metrics_collector.log_info("No S3 partitions found", backfill_start_date=self.config.backfill_start_date, backfill_end_date=self.config.backfill_end_date, s3_bucket=self.config.s3_bucket, note="Ensure data has been exported to S3 using EXPORT_OWNED_CLIP_ANALYSIS_TO_S3" ) return partition_paths def _load_and_prepare_data(self, partition_paths: List[str]) -> DataFrame: """Load data from S3 and prepare for processing.""" # Load S3 data dataframe = self.s3_data_loader.load_data_from_partitions( partition_paths, self.config.backfill_end_date ) # Check if we have data self.total_rows = dataframe.count() self.metrics_collector.set_total_rows(self.total_rows) self.metrics_collector.log_checkpoint("S3_DATA_LOADED", total_rows=self.total_rows, total_rows_formatted=f"{self.total_rows:,}", partitions_processed=len(partition_paths) ) if self.total_rows == 0: return None # Optimize DataFrame for processing optimized_df = self.s3_data_loader.optimize_dataframe_for_processing(dataframe) return optimized_df def _process_data_parallel(self, dataframe: DataFrame): """Process data using parallel Spark partitions.""" self.metrics_collector.log_checkpoint("PARALLEL_PROCESSING_STARTING", total_rows=self.total_rows, spark_partitions=dataframe.rdd.getNumPartitions() ) # Create serializable partition processor config_dict = self.config.to_dict() partition_processor = create_partition_processor(config_dict) # Process each partition using the serializable function dataframe.foreachPartition(partition_processor) self.metrics_collector.log_checkpoint("PARALLEL_PROCESSING_COMPLETE") def _complete_job_successfully(self) -> Dict[str, Any]: """Complete job successfully and return results.""" final_progress = self.metrics_collector.get_progress_info() self.metrics_collector.log_checkpoint("JOB_COMPLETED_SUCCESS", processed_rows=final_progress["processed_rows"], successful_inserts=final_progress["successful_inserts"], batches_processed=final_progress["batches_processed"], total_duration=final_progress["elapsed_seconds"], average_throughput_rows_per_sec=final_progress["throughput_rows_per_sec"] ) return { "status": "SUCCESS", "job_start_time": self.job_start_time.isoformat(), "job_end_time": datetime.now().isoformat(), "progress": final_progress, "job_type": "S3_TO_RDS_BACKFILL" } def _handle_no_data(self) -> Dict[str, Any]: """Handle case where no data is found.""" self.metrics_collector.log_info("No data found for processing", backfill_start_date=self.config.backfill_start_date, backfill_end_date=self.config.backfill_end_date ) return { "status": "NO_DATA", "message": "No data found in the specified date range", "job_start_time": self.job_start_time.isoformat(), "job_end_time": datetime.now().isoformat() } def _handle_job_failure(self, error: Exception) -> Dict[str, Any]: """Handle job failure with comprehensive error logging.""" error_progress = self.metrics_collector.get_progress_info() self.metrics_collector.log_error("Job execution failed", error_type=type(error).__name__, error_message=str(error)[:500], processed_rows=error_progress.get("processed_rows", 0), successful_inserts=error_progress.get("successful_inserts", 0), elapsed_seconds=error_progress.get("elapsed_seconds", 0) ) return { "status": "FAILED", "error_type": type(error).__name__, "error_message": str(error), "job_start_time": self.job_start_time.isoformat(), "job_end_time": datetime.now().isoformat(), "progress_at_failure": error_progress } def _cleanup_resources(self): """Clean up all resources and connections.""" try: self.database_manager.close_all_connections() self.metrics_collector.log_info("Resource cleanup completed") except Exception as e: self.metrics_collector.log_warning("Error during resource cleanup", error=str(e)[:200] )