# Spark metrics collection and analysis
from pyspark.sql import SparkSession
import json
import time

class SparkMetricsCollector:
    def __init__(self, spark_session):
        self.spark = spark_session
        self.sc = spark_session.sparkContext
        
    def collect_job_metrics(self):
        """Collect job metrics"""
        status_tracker = self.sc.statusTracker()
        
        # Active job information
        active_jobs = status_tracker.getActiveJobIds()
        
        # Collect metrics by job
        job_metrics = {}
        for job_id in active_jobs:
            job_info = status_tracker.getJobInfo(job_id)
            job_metrics[job_id] = {
                'status': job_info.status,
                'num_tasks': job_info.numTasks,
                'num_active_tasks': job_info.numActiveTasks,
                'num_completed_tasks': job_info.numCompletedTasks,
                'num_failed_tasks': job_info.numFailedTasks
            }
        
        return job_metrics
    
    def collect_stage_metrics(self):
        """Collect stage metrics"""
        status_tracker = self.sc.statusTracker()
        
        # Active stage information
        active_stages = status_tracker.getActiveStageIds()
        
        stage_metrics = {}
        for stage_id in active_stages:
            stage_info = status_tracker.getStageInfo(stage_id)
            stage_metrics[stage_id] = {
                'num_tasks': stage_info.numTasks,
                'num_active_tasks': stage_info.numActiveTasks,
                'num_completed_tasks': stage_info.numCompletedTasks,
                'num_failed_tasks': stage_info.numFailedTasks,
                'executor_run_time': stage_info.executorRunTime,
                'executor_cpu_time': stage_info.executorCpuTime,
                'input_bytes': stage_info.inputBytes,
                'output_bytes': stage_info.outputBytes,
                'shuffle_read_bytes': stage_info.shuffleReadBytes,
                'shuffle_write_bytes': stage_info.shuffleWriteBytes
            }
        
        return stage_metrics
    
    def collect_executor_metrics(self):
        """Collect executor metrics"""
        status_tracker = self.sc.statusTracker()
        executor_infos = status_tracker.getExecutorInfos()
        
        executor_metrics = {}
        for executor_info in executor_infos:
            executor_metrics[executor_info.executorId] = {
                'host': executor_info.host,
                'total_cores': executor_info.totalCores,
                'max_memory': executor_info.maxMemory,
                'memory_used': executor_info.memoryUsed,
                'disk_used': executor_info.diskUsed,
                'active_tasks': executor_info.activeTasks,
                'completed_tasks': executor_info.completedTasks,
                'failed_tasks': executor_info.failedTasks,
                'total_duration': executor_info.totalDuration,
                'total_gc_time': executor_info.totalGCTime
            }
        
        return executor_metrics

# Usage example
def monitor_spark_application():
    spark = SparkSession.builder.appName("MetricsExample").getOrCreate()
    collector = SparkMetricsCollector(spark)
    
    # Collect metrics periodically
    while True:
        job_metrics = collector.collect_job_metrics()
        stage_metrics = collector.collect_stage_metrics()
        executor_metrics = collector.collect_executor_metrics()
        
        print("=== Job Metrics ===")
        print(json.dumps(job_metrics, indent=2))
        
        print("=== Stage Metrics ===")
        print(json.dumps(stage_metrics, indent=2))
        
        print("=== Executor Metrics ===")
        print(json.dumps(executor_metrics, indent=2))
        
        time.sleep(10)  # Collect every 10 seconds

# Performance profiling class
class SparkProfiler:
    def __init__(self, spark_session):
        self.spark = spark_session
        self.sc = spark_session.sparkContext
    
    def profile_query_execution(self, query_name, df):
        """Profile query execution"""
        import time
        
        start_time = time.time()
        
        # Analyze execution plan
        execution_plan = df.explain(True)
        
        # Execute query
        result = df.collect()
        
        end_time = time.time()
        execution_time = end_time - start_time
        
        # Collect metrics
        status_tracker = self.sc.statusTracker()
        executor_infos = status_tracker.getExecutorInfos()
        
        total_memory_used = sum(info.memoryUsed for info in executor_infos)
        total_gc_time = sum(info.totalGCTime for info in executor_infos)
        
        profile_result = {
            'query_name': query_name,
            'execution_time': execution_time,
            'total_memory_used': total_memory_used,
            'total_gc_time': total_gc_time,
            'execution_plan': execution_plan,
            'result_count': len(result)
        }
        
        return profile_result
    
    def analyze_data_skew(self, df, key_columns):
        """Analyze data skew"""
        # Analyze data distribution by key
        key_counts = df.groupBy(*key_columns).count()
        
        # Collect statistics
        stats = key_counts.select(
            count("*").alias("total_keys"),
            min("count").alias("min_count"),
            max("count").alias("max_count"),
            avg("count").alias("avg_count"),
            stddev("count").alias("stddev_count")
        ).collect()[0]
        
        # Calculate skew ratio
        skew_ratio = stats['max_count'] / stats['avg_count'] if stats['avg_count'] > 0 else 0
        
        skew_analysis = {
            'total_keys': stats['total_keys'],
            'min_count': stats['min_count'],
            'max_count': stats['max_count'],
            'avg_count': stats['avg_count'],
            'stddev_count': stats['stddev_count'],
            'skew_ratio': skew_ratio,
            'is_skewed': skew_ratio > 2.0  # Skew threshold
        }
        
        return skew_analysis
    
    def monitor_memory_usage(self):
        """Monitor memory usage"""
        status_tracker = self.sc.statusTracker()
        executor_infos = status_tracker.getExecutorInfos()
        
        memory_stats = {
            'total_executors': len(executor_infos),
            'total_memory_used': sum(info.memoryUsed for info in executor_infos),
            'total_max_memory': sum(info.maxMemory for info in executor_infos),
            'memory_utilization': 0,
            'executor_details': []
        }
        
        for info in executor_infos:
            executor_detail = {
                'executor_id': info.executorId,
                'host': info.host,
                'memory_used': info.memoryUsed,
                'max_memory': info.maxMemory,
                'utilization': info.memoryUsed / info.maxMemory if info.maxMemory > 0 else 0
            }
            memory_stats['executor_details'].append(executor_detail)
        
        if memory_stats['total_max_memory'] > 0:
            memory_stats['memory_utilization'] = (
                memory_stats['total_memory_used'] / memory_stats['total_max_memory']
            )
        
        return memory_stats

# Performance analysis tool usage example
def performance_analysis_example():
    spark = SparkSession.builder.appName("PerformanceAnalysis").getOrCreate()
    profiler = SparkProfiler(spark)
    
    # Create sample data
    data = [(i, f"user_{i}", i * 10) for i in range(10000)]
    df = spark.createDataFrame(data, ["id", "name", "value"])
    
    # Query profiling
    profile_result = profiler.profile_query_execution(
        "sample_aggregation",
        df.groupBy("name").agg(sum("value").alias("total_value"))
    )
    
    print("Query Profile Result:")
    print(json.dumps(profile_result, indent=2))
    
    # Data skew analysis
    skew_analysis = profiler.analyze_data_skew(df, ["name"])
    print("\nData Skew Analysis:")
    print(json.dumps(skew_analysis, indent=2))
    
    # Memory usage monitoring
    memory_stats = profiler.monitor_memory_usage()
    print("\nMemory Usage Stats:")
    print(json.dumps(memory_stats, indent=2))

# Memory optimization tools
class MemoryOptimizer:
    def __init__(self, spark_session):
        self.spark = spark_session
        
    def optimize_dataframe_memory(self, df):
        """Optimize DataFrame memory"""
        from pyspark.sql.types import IntegerType, LongType, FloatType, DoubleType
        
        optimized_df = df
        
        # Optimize column types
        for field in df.schema.fields:
            field_name = field.name
            field_type = field.dataType
            
            if isinstance(field_type, IntegerType):
                # Change to smaller type if no large integer values
                max_val = df.select(max(col(field_name))).collect()[0][0]
                min_val = df.select(min(col(field_name))).collect()[0][0]
                
                if -128 <= min_val <= 127 and -128 <= max_val <= 127:
                    optimized_df = optimized_df.withColumn(
                        field_name, col(field_name).cast("tinyint")
                    )
                elif -32768 <= min_val <= 32767 and -32768 <= max_val <= 32767:
                    optimized_df = optimized_df.withColumn(
                        field_name, col(field_name).cast("smallint")
                    )
            
            elif isinstance(field_type, FloatType):
                # Change Float to Double for better precision
                optimized_df = optimized_df.withColumn(
                    field_name, col(field_name).cast("double")
                )
        
        return optimized_df
    
    def calculate_memory_usage(self, df):
        """Calculate memory usage"""
        # Estimate DataFrame size
        num_rows = df.count()
        num_cols = len(df.columns)
        
        # Estimate type size by column
        type_sizes = {
            'string': 50,  # Average string length
            'int': 4,
            'bigint': 8,
            'double': 8,
            'float': 4,
            'boolean': 1,
            'date': 8,
            'timestamp': 8
        }
        
        estimated_size = 0
        for field in df.schema.fields:
            field_type = str(field.dataType)
            base_type = field_type.split('(')[0].lower()
            
            if base_type in type_sizes:
                estimated_size += type_sizes[base_type]
            else:
                estimated_size += 8  # Default value
        
        total_estimated_size = num_rows * num_cols * estimated_size
        
        return {
            'num_rows': num_rows,
            'num_cols': num_cols,
            'estimated_size_bytes': total_estimated_size,
            'estimated_size_mb': total_estimated_size / (1024 * 1024)
        }
    
    def optimize_caching_strategy(self, df, access_pattern):
        """Optimize caching strategy"""
        from pyspark import StorageLevel
        
        memory_stats = self.calculate_memory_usage(df)
        size_mb = memory_stats['estimated_size_mb']
        
        recommendations = []
        
        if access_pattern['frequency'] == 'high' and size_mb < 1000:
            recommendations.append({
                'strategy': 'MEMORY_ONLY',
                'reason': 'Frequently used and small data'
            })
        elif access_pattern['frequency'] == 'high' and size_mb >= 1000:
            recommendations.append({
                'strategy': 'MEMORY_AND_DISK_SER',
                'reason': 'Frequently used but large data'
            })
        elif access_pattern['frequency'] == 'medium':
            recommendations.append({
                'strategy': 'DISK_ONLY',
                'reason': 'Medium frequency usage data'
            })
        else:
            recommendations.append({
                'strategy': 'NO_CACHING',
                'reason': 'Low frequency usage data'
            })
        
        return {
            'data_size_mb': size_mb,
            'access_frequency': access_pattern['frequency'],
            'recommendations': recommendations
        }

# Memory optimization example
def memory_optimization_example():
    spark = SparkSession.builder.appName("MemoryOptimization").getOrCreate()
    optimizer = MemoryOptimizer(spark)
    
    # Create sample data
    data = [(i, f"user_{i}", i * 1.5, i % 2 == 0) for i in range(100000)]
    df = spark.createDataFrame(data, ["id", "name", "score", "is_active"])
    
    # Memory optimization
    optimized_df = optimizer.optimize_dataframe_memory(df)
    
    # Calculate memory usage
    original_memory = optimizer.calculate_memory_usage(df)
    optimized_memory = optimizer.calculate_memory_usage(optimized_df)
    
    print("Original Memory Usage:")
    print(json.dumps(original_memory, indent=2))
    
    print("\nOptimized Memory Usage:")
    print(json.dumps(optimized_memory, indent=2))
    
    # Optimize caching strategy
    access_pattern = {'frequency': 'high'}
    caching_strategy = optimizer.optimize_caching_strategy(df, access_pattern)
    
    print("\nCaching Strategy:")
    print(json.dumps(caching_strategy, indent=2))

# Cluster tuning tools
class ClusterTuner:
    def __init__(self, spark_session):
        self.spark = spark_session
        
    def analyze_cluster_resources(self):
        """Analyze cluster resources"""
        status_tracker = self.spark.sparkContext.statusTracker()
        executor_infos = status_tracker.getExecutorInfos()
        
        cluster_analysis = {
            'total_executors': len(executor_infos),
            'total_cores': sum(info.totalCores for info in executor_infos),
            'total_memory': sum(info.maxMemory for info in executor_infos),
            'memory_utilization': 0,
            'cpu_utilization': 0,
            'resource_recommendations': []
        }
        
        total_memory_used = sum(info.memoryUsed for info in executor_infos)
        if cluster_analysis['total_memory'] > 0:
            cluster_analysis['memory_utilization'] = total_memory_used / cluster_analysis['total_memory']
        
        # Resource optimization recommendations
        if cluster_analysis['memory_utilization'] > 0.8:
            cluster_analysis['resource_recommendations'].append(
                "High memory usage. Consider allocating more memory"
            )
        
        if cluster_analysis['total_executors'] < 3:
            cluster_analysis['resource_recommendations'].append(
                "Low number of executors. Consider more executors for scalability"
            )
        
        return cluster_analysis
    
    def optimize_spark_config(self, workload_type):
        """Optimize Spark configuration by workload type"""
        config_recommendations = {}
        
        if workload_type == "batch":
            config_recommendations = {
                "spark.sql.adaptive.enabled": "true",
                "spark.sql.adaptive.coalescePartitions.enabled": "true",
                "spark.sql.adaptive.advisoryPartitionSizeInBytes": "128MB",
                "spark.serializer": "org.apache.spark.serializer.KryoSerializer",
                "spark.sql.execution.arrow.pyspark.enabled": "true"
            }
        elif workload_type == "streaming":
            config_recommendations = {
                "spark.sql.streaming.checkpointLocation": "/tmp/checkpoint",
                "spark.sql.streaming.stateStore.providerClass": "org.apache.spark.sql.execution.streaming.state.HDFSBackedStateStoreProvider",
                "spark.sql.streaming.minBatchesToRetain": "2",
                "spark.sql.streaming.stateStore.maintenanceInterval": "60s"
            }
        elif workload_type == "ml":
            config_recommendations = {
                "spark.sql.execution.arrow.maxRecordsPerBatch": "10000",
                "spark.serializer": "org.apache.spark.serializer.KryoSerializer",
                "spark.sql.adaptive.enabled": "true",
                "spark.sql.adaptive.coalescePartitions.enabled": "true"
            }
        
        return config_recommendations
    
    def calculate_optimal_partitions(self, data_size_gb, executor_cores, num_executors):
        """Calculate optimal number of partitions"""
        total_cores = executor_cores * num_executors
        
        # General recommendation: 2-3 partitions per core
        min_partitions = total_cores * 2
        max_partitions = total_cores * 3
        
        # Adjust based on data size
        target_partition_size_mb = 128
        size_based_partitions = int(data_size_gb * 1024 / target_partition_size_mb)
        
        optimal_partitions = max(min_partitions, min(max_partitions, size_based_partitions))
        
        return {
            'total_cores': total_cores,
            'min_partitions': min_partitions,
            'max_partitions': max_partitions,
            'size_based_partitions': size_based_partitions,
            'optimal_partitions': optimal_partitions,
            'recommendation': f"Recommended partitions: {optimal_partitions}"
        }

# Cluster tuning example
def cluster_tuning_example():
    spark = SparkSession.builder.appName("ClusterTuning").getOrCreate()
    tuner = ClusterTuner(spark)
    
    # Analyze cluster resources
    cluster_analysis = tuner.analyze_cluster_resources()
    print("Cluster Resource Analysis:")
    print(json.dumps(cluster_analysis, indent=2))
    
    # Optimize configuration by workload
    batch_config = tuner.optimize_spark_config("batch")
    print("\nBatch Workload Configuration:")
    print(json.dumps(batch_config, indent=2))
    
    # Calculate optimal partitions
    partition_analysis = tuner.calculate_optimal_partitions(
        data_size_gb=10,
        executor_cores=4,
        num_executors=5
    )
    print("\nPartition Analysis:")
    print(json.dumps(partition_analysis, indent=2))

# Performance monitoring system
class PerformanceMonitoringSystem:
    def __init__(self, spark_session):
        self.spark = spark_session
        self.metrics_history = []
        self.alerts = []
        
    def collect_comprehensive_metrics(self):
        """Collect comprehensive metrics"""
        status_tracker = self.spark.sparkContext.statusTracker()
        
        # Basic metrics
        executor_infos = status_tracker.getExecutorInfos()
        active_jobs = status_tracker.getActiveJobIds()
        active_stages = status_tracker.getActiveStageIds()
        
        # System metrics
        system_metrics = {
            'timestamp': time.time(),
            'active_jobs_count': len(active_jobs),
            'active_stages_count': len(active_stages),
            'total_executors': len(executor_infos),
            'total_cores': sum(info.totalCores for info in executor_infos),
            'total_memory': sum(info.maxMemory for info in executor_infos),
            'used_memory': sum(info.memoryUsed for info in executor_infos),
            'total_gc_time': sum(info.totalGCTime for info in executor_infos),
            'total_tasks': sum(info.completedTasks + info.failedTasks for info in executor_infos),
            'failed_tasks': sum(info.failedTasks for info in executor_infos)
        }
        
        # Add to metrics history
        self.metrics_history.append(system_metrics)
        
        # Keep only recent 100
        if len(self.metrics_history) > 100:
            self.metrics_history = self.metrics_history[-100:]
        
        return system_metrics
    
    def analyze_performance_trends(self):
        """Analyze performance trends"""
        if len(self.metrics_history) < 2:
            return {"message": "Insufficient data"}
        
        recent_metrics = self.metrics_history[-10:]  # Recent 10
        
        # Calculate trends
        memory_trend = self._calculate_trend([m['used_memory'] for m in recent_metrics])
        gc_trend = self._calculate_trend([m['total_gc_time'] for m in recent_metrics])
        failure_trend = self._calculate_trend([m['failed_tasks'] for m in recent_metrics])
        
        return {
            'memory_trend': memory_trend,
            'gc_trend': gc_trend,
            'failure_trend': failure_trend,
            'analysis_period': f"{len(recent_metrics)} samples",
            'recommendations': self._generate_recommendations(memory_trend, gc_trend, failure_trend)
        }
    
    def _calculate_trend(self, values):
        """Calculate trend (linear regression based)"""
        if len(values) < 2:
            return "insufficient_data"
        
        # Simple trend calculation
        first_half = values[:len(values)//2]
        second_half = values[len(values)//2:]
        
        first_avg = sum(first_half) / len(first_half)
        second_avg = sum(second_half) / len(second_half)
        
        if second_avg > first_avg * 1.1:
            return "increasing"
        elif second_avg < first_avg * 0.9:
            return "decreasing"
        else:
            return "stable"
    
    def _generate_recommendations(self, memory_trend, gc_trend, failure_trend):
        """Generate recommendations"""
        recommendations = []
        
        if memory_trend == "increasing":
            recommendations.append("Memory usage is increasing. Consider memory optimization")
        
        if gc_trend == "increasing":
            recommendations.append("GC time is increasing. Consider JVM tuning")
        
        if failure_trend == "increasing":
            recommendations.append("Task failures are increasing. Check resource allocation")
        
        return recommendations
    
    def check_alerts(self, current_metrics):
        """Check alerts"""
        alerts = []
        
        # Memory usage alert
        memory_utilization = current_metrics['used_memory'] / current_metrics['total_memory']
        if memory_utilization > 0.9:
            alerts.append({
                'type': 'warning',
                'message': f'High memory usage: {memory_utilization:.2%}',
                'timestamp': current_metrics['timestamp']
            })
        
        # Task failure rate alert
        total_tasks = current_metrics['total_tasks']
        failed_tasks = current_metrics['failed_tasks']
        if total_tasks > 0 and failed_tasks / total_tasks > 0.1:
            alerts.append({
                'type': 'error',
                'message': f'High task failure rate: {failed_tasks/total_tasks:.2%}',
                'timestamp': current_metrics['timestamp']
            })
        
        # GC time alert
        if current_metrics['total_gc_time'] > 30000:  # 30 seconds
            alerts.append({
                'type': 'warning',
                'message': f'Long GC time: {current_metrics["total_gc_time"]}ms',
                'timestamp': current_metrics['timestamp']
            })
        
        self.alerts.extend(alerts)
        return alerts

# Performance monitoring system example
def performance_monitoring_example():
    spark = SparkSession.builder.appName("PerformanceMonitoring").getOrCreate()
    monitoring_system = PerformanceMonitoringSystem(spark)
    
    # Execute sample workload
    data = [(i, f"category_{i % 100}", i * 10) for i in range(50000)]
    df = spark.createDataFrame(data, ["id", "category", "value"])
    
    # Collect metrics
    for i in range(5):
        metrics = monitoring_system.collect_comprehensive_metrics()
        alerts = monitoring_system.check_alerts(metrics)
        
        print(f"\n=== Metrics Collection {i+1} ===")
        print(f"Memory Usage: {metrics['used_memory']:,} / {metrics['total_memory']:,}")
        print(f"Active Jobs: {metrics['active_jobs_count']}")
        print(f"Failed Tasks: {metrics['failed_tasks']}")
        
        if alerts:
            print("Alerts:")
            for alert in alerts:
                print(f"  [{alert['type'].upper()}] {alert['message']}")
        
        time.sleep(2)
    
    # Analyze performance trends
    trend_analysis = monitoring_system.analyze_performance_trends()
    print("\n=== Performance Trend Analysis ===")
    print(json.dumps(trend_analysis, indent=2))

# Production monitoring dashboard
class ProductionDashboard:
    def __init__(self, spark_session):
        self.spark = spark_session
        self.alert_thresholds = {
            'memory_usage': 0.85,
            'gc_time': 30000,
            'failed_tasks_ratio': 0.05,
            'execution_time': 3600
        }
        
    def generate_health_report(self):
        """Generate system health report"""
        status_tracker = self.spark.sparkContext.statusTracker()
        executor_infos = status_tracker.getExecutorInfos()
        
        # Collect basic metrics
        total_memory = sum(info.maxMemory for info in executor_infos)
        used_memory = sum(info.memoryUsed for info in executor_infos)
        total_gc_time = sum(info.totalGCTime for info in executor_infos)
        total_tasks = sum(info.completedTasks + info.failedTasks for info in executor_infos)
        failed_tasks = sum(info.failedTasks for info in executor_infos)
        
        # Calculate health scores
        memory_score = 100 - (used_memory / total_memory * 100) if total_memory > 0 else 100
        gc_score = max(0, 100 - (total_gc_time / 60000 * 100))  # GC time per minute
        reliability_score = 100 - (failed_tasks / total_tasks * 100) if total_tasks > 0 else 100
        
        overall_health = (memory_score + gc_score + reliability_score) / 3
        
        return {
            'timestamp': time.time(),
            'overall_health_score': overall_health,
            'memory_health': memory_score,
            'gc_health': gc_score,
            'reliability_health': reliability_score,
            'alerts': self._check_production_alerts(
                used_memory, total_memory, total_gc_time, failed_tasks, total_tasks
            ),
            'recommendations': self._generate_production_recommendations(
                memory_score, gc_score, reliability_score
            )
        }
    
    def _check_production_alerts(self, used_memory, total_memory, gc_time, failed_tasks, total_tasks):
        """Check production alerts"""
        alerts = []
        
        # Memory usage alert
        memory_ratio = used_memory / total_memory if total_memory > 0 else 0
        if memory_ratio > self.alert_thresholds['memory_usage']:
            alerts.append({
                'level': 'critical',
                'type': 'memory_usage',
                'message': f'Memory usage exceeded threshold: {memory_ratio:.2%}',
                'value': memory_ratio,
                'threshold': self.alert_thresholds['memory_usage']
            })
        
        # GC time alert
        if gc_time > self.alert_thresholds['gc_time']:
            alerts.append({
                'level': 'warning',
                'type': 'gc_time',
                'message': f'GC time exceeded threshold: {gc_time}ms',
                'value': gc_time,
                'threshold': self.alert_thresholds['gc_time']
            })
        
        # Task failure rate alert
        failure_ratio = failed_tasks / total_tasks if total_tasks > 0 else 0
        if failure_ratio > self.alert_thresholds['failed_tasks_ratio']:
            alerts.append({
                'level': 'critical',
                'type': 'task_failure',
                'message': f'Task failure rate exceeded threshold: {failure_ratio:.2%}',
                'value': failure_ratio,
                'threshold': self.alert_thresholds['failed_tasks_ratio']
            })
        
        return alerts
    
    def _generate_production_recommendations(self, memory_score, gc_score, reliability_score):
        """Generate production recommendations"""
        recommendations = []
        
        if memory_score < 70:
            recommendations.append({
                'category': 'memory',
                'priority': 'high',
                'action': 'Increase memory allocation or optimize memory usage',
                'details': 'Adjust executor-memory or executor-memoryFraction settings'
            })
        
        if gc_score < 70:
            recommendations.append({
                'category': 'gc',
                'priority': 'medium',
                'action': 'Optimize JVM GC settings',
                'details': 'Consider -XX:+UseG1GC, -XX:MaxGCPauseMillis settings'
            })
        
        if reliability_score < 90:
            recommendations.append({
                'category': 'reliability',
                'priority': 'high',
                'action': 'Analyze and resolve task failure causes',
                'details': 'Log analysis and resource allocation check required'
            })
        
        return recommendations
    
    def export_metrics_to_external_system(self, health_report):
        """Export metrics to external system"""
        # Send metrics to Prometheus, InfluxDB, Elasticsearch, etc.
        metrics_data = {
            'spark_health_score': health_report['overall_health_score'],
            'memory_health': health_report['memory_health'],
            'gc_health': health_report['gc_health'],
            'reliability_health': health_report['reliability_health'],
            'timestamp': health_report['timestamp']
        }
        
        # In real environment, implement HTTP request or database storage logic
        print("Metrics exported to external system:")
        print(json.dumps(metrics_data, indent=2))
        
        return True

# Production dashboard example
def production_dashboard_example():
    spark = SparkSession.builder.appName("ProductionDashboard").getOrCreate()
    dashboard = ProductionDashboard(spark)
    
    # Generate system health report
    health_report = dashboard.generate_health_report()
    
    print("=== Production Health Report ===")
    print(f"Overall Health Score: {health_report['overall_health_score']:.1f}/100")
    print(f"Memory Health: {health_report['memory_health']:.1f}/100")
    print(f"GC Health: {health_report['gc_health']:.1f}/100")
    print(f"Reliability Health: {health_report['reliability_health']:.1f}/100")
    
    # Check alerts
    if health_report['alerts']:
        print("\n=== Alerts ===")
        for alert in health_report['alerts']:
            print(f"[{alert['level'].upper()}] {alert['message']}")
    
    # Check recommendations
    if health_report['recommendations']:
        print("\n=== Recommendations ===")
        for rec in health_report['recommendations']:
            print(f"[{rec['priority'].upper()}] {rec['action']}")
            print(f"  Details: {rec['details']}")
    
    # Export metrics
    dashboard.export_metrics_to_external_system(health_report)
    
    return health_report