from pyspark.sql import SparkSession

# Configure Spark with monitoring
# Parameter: spark.eventLog.enabled — enables event logging
# Parameter: spark.eventLog.dir — directory for event logs
spark = SparkSession.builder \
    .appName("MonitoringExample") \
    .config("spark.eventLog.enabled", "true") \
    .config("spark.eventLog.dir", "hdfs:///spark-history") \
    .config("spark.yarn.historyServer.address", "history-server-host:18080") \
    .config("spark.ui.port", "4040") \
    .config("spark.ui.enabled", "true") \
    .config("spark.ui.killEnabled", "true") \
    .getOrCreate()

# Read data
df = spark.read \
    .format("parquet") \
    .load("hdfs:///data/input")

# Process data with monitoring
result = df \
    .repartition(100) \
    .groupBy("category") \
    .agg({"value": "sum", "count": "count"})

# Write results
result.write \
    .mode("overwrite") \
    .parquet("hdfs:///data/output")

# Access Spark UI
print(f"Spark UI available at: http://localhost:4040")

spark.stop()

from pyspark.sql import SparkSession

# Configure Spark with Prometheus metrics
# Parameter: spark.metrics.conf — metrics configuration file
# Parameter: spark.metrics.staticConf.enabled — enable static metrics
spark = SparkSession.builder \
    .appName("PrometheusMonitoring") \
    .config("spark.metrics.conf", "metrics.properties") \
    .config("spark.metrics.staticConf.enabled", "true") \
    .config("spark.metrics.namespace", "spark_application") \
    .getOrCreate()

# Create metrics configuration file
metrics_config = """
# Prometheus metrics configuration
*.sink.prometheus.class=org.apache.spark.metrics.sink.PrometheusServletSink
*.sink.prometheus.port=9090

# Application metrics
appapplicationmaster.sink.prometheus.class=org.apache.spark.metrics.sink.PrometheusServletSink
appapplicationmaster.sink.prometheus.path=/metrics/prometheus

# Executor metrics
executor.sink.prometheus.class=org.apache.spark.metrics.sink.PrometheusServletSink
executor.sink.prometheus.path=/metrics/prometheus

# JVM metrics
jvm.sink.prometheus.class=org.apache.spark.metrics.sink.PrometheusServletSink
jvm.sink.prometheus.path=/metrics/prometheus
"""

# Write metrics configuration
with open("metrics.properties", "w") as f:
    f.write(metrics_config)

# Read data
df = spark.read \
    .format("parquet") \
    .load("hdfs:///data/input")

# Process data
result = df.groupBy("category").count()

# Write results
result.write \
    .mode("overwrite") \
    .parquet("hdfs:///data/output")

# Prometheus metrics will be available at:
# http://executor-host:9090/metrics/prometheus

spark.stop()

from pyspark.sql import SparkSession
import time
import threading

# Configure Spark with JMX metrics
spark = SparkSession.builder \
    .appName("JMXMonitoring") \
    .config("spark.metrics.conf", "metrics.properties") \
    .getOrCreate()

# Create JMX metrics configuration
jmx_config = """
# JMX metrics configuration
*.sink.jmx.class=org.apache.spark.metrics.sink.JmxSink
*.sink.jmx.port=9999

# Application metrics
appapplicationmaster.sink.jmx.class=org.apache.spark.metrics.sink.JmxSink

# Executor metrics
executor.sink.jmx.class=org.apache.spark.metrics.sink.JmxSink

# Shuffle metrics
shuffle.sink.jmx.class=org.apache.spark.metrics.sink.JmxSink
"""

# Write JMX configuration
with open("metrics.properties", "w") as f:
    f.write(jmx_config)

# Function to collect custom metrics
def collect_custom_metrics():
    """Collect custom application metrics"""
    sc = spark.sparkContext
    
    # Get executor information
    executor_info = sc._jsc.sc().statusTracker().getExecutorInfos()
    
    # Log executor count
    print(f"Number of executors: {len(executor_info)}")
    
    # Monitor processing
    df = spark.read \
        .format("parquet") \
        .load("hdfs:///data/input")
    
    # Cache and count
    df.cache()
    count = df.count()
    
    print(f"Row count: {count}")
    
    # Process data
    result = df \
        .repartition(100) \
        .groupBy("category") \
        .agg({"value": "sum"})
    
    result.write \
        .mode("overwrite") \
        .parquet("hdfs:///data/output")

# Run metrics collection
collect_custom_metrics()

spark.stop()

from pyspark.sql import SparkSession
import json
import logging
from datetime import datetime

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class SparkMonitor:
    def __init__(self, app_name, config):
        """Initialize Spark monitoring"""
        self.app_name = app_name
        self.config = config
        self.spark = self.create_spark_session()
        self.metrics = {}
    
    def create_spark_session(self):
        """Create Spark session with monitoring configuration"""
        return SparkSession.builder \
            .appName(self.app_name) \
            .config("spark.eventLog.enabled", "true") \
            .config("spark.eventLog.dir", self.config["event_log_dir"]) \
            .config("spark.yarn.historyServer.address", self.config["history_server"]) \
            .config("spark.metrics.conf", "metrics.properties") \
            .config("spark.ui.port", "4040") \
            .getOrCreate()
    
    def collect_metrics(self):
        """Collect application metrics"""
        sc = self.spark.sparkContext
        
        # Get executor information
        executor_info = sc._jsc.sc().statusTracker().getExecutorInfos()
        
        self.metrics = {
            "timestamp": datetime.now().isoformat(),
            "app_name": self.app_name,
            "num_executors": len(executor_info),
            "application_id": sc.applicationId,
            "spark_version": sc.version
        }
        
        return self.metrics
    
    def monitor_processing(self, input_path, output_path):
        """Monitor data processing"""
        start_time = time.time()
        
        # Read data
        df = self.spark.read \
            .format("parquet") \
            .load(input_path)
        
        # Cache and count
        df.cache()
        count = df.count()
        
        # Process data
        result = df \
            .repartition(100) \
            .groupBy("category") \
            .agg({"value": "sum", "count": "count"})
        
        # Write results
        result.write \
            .mode("overwrite") \
            .parquet(output_path)
        
        end_time = time.time()
        
        # Update metrics
        self.metrics.update({
            "input_rows": count,
            "processing_time": end_time - start_time,
            "output_path": output_path,
            "status": "completed"
        })
        
        return self.metrics
    
    def generate_report(self):
        """Generate monitoring report"""
        report = {
            "summary": self.metrics,
            "recommendations": self.generate_recommendations()
        }
        
        # Save report
        with open(f"monitoring_report_{self.app_name}.json", "w") as f:
            json.dump(report, f, indent=2)
        
        return report
    
    def generate_recommendations(self):
        """Generate optimization recommendations"""
        recommendations = []
        
        # Analyze metrics and generate recommendations
        if self.metrics.get("processing_time", 0) > 3600:
            recommendations.append("Consider increasing parallelism or partitioning data differently")
        
        if self.metrics.get("num_executors", 0) < 5:
            recommendations.append("Consider adding more executors for better parallelism")
        
        return recommendations

# Example usage
if __name__ == "__main__":
    # Configuration
    config = {
        "event_log_dir": "hdfs:///spark-history",
        "history_server": "history-server-host:18080"
    }
    
    # Initialize monitor
    monitor = SparkMonitor("MonitoredApplication", config)
    
    # Collect metrics
    metrics = monitor.collect_metrics()
    print(f"Initial metrics: {metrics}")
    
    # Monitor processing
    processing_metrics = monitor.monitor_processing(
        "hdfs:///data/input",
        "hdfs:///data/output"
    )
    print(f"Processing metrics: {processing_metrics}")
    
    # Generate report
    report = monitor.generate_report()
    print(f"Monitoring report saved")
    
    monitor.spark.stop()