from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *
import pandas as pd
import numpy as np

class PySparkIntroduction:
    """PySpark介紹與優(yōu)勢(shì)分析"""
    
    def __init__(self):
        self.advantages = {
            "性能優(yōu)勢(shì)": [
                "內(nèi)存計(jì)算比Hadoop MapReduce快100倍",
                "基于DAG的優(yōu)化執(zhí)行引擎",
                "懶加載機(jī)制優(yōu)化計(jì)算流程"
            ],
            "易用性?xún)?yōu)勢(shì)": [
                "Python簡(jiǎn)潔的API接口",
                "與Pandas無(wú)縫集成",
                "豐富的機(jī)器學(xué)習(xí)庫(kù)"
            ],
            "生態(tài)系統(tǒng)": [
                "Spark SQL: 結(jié)構(gòu)化數(shù)據(jù)處理",
                "Spark Streaming: 實(shí)時(shí)數(shù)據(jù)處理", 
                "MLlib: 機(jī)器學(xué)習(xí)庫(kù)",
                "GraphX: 圖計(jì)算"
            ]
        }
    
    def demonstrate_performance_comparison(self):
        """展示性能對(duì)比"""
        data_sizes = [1, 10, 100, 1000]  # GB
        pandas_times = [1, 15, 180, 1800]  # 秒
        pyspark_times = [2, 5, 20, 120]    # 秒
        
        comparison_data = {
            "數(shù)據(jù)大小(GB)": data_sizes,
            "Pandas處理時(shí)間(秒)": pandas_times,
            "PySpark處理時(shí)間(秒)": pyspark_times,
            "性能提升倍數(shù)": [p/t for p, t in zip(pandas_times, pyspark_times)]
        }
        
        df = pd.DataFrame(comparison_data)
        return df
    
    def spark_architecture_overview(self):
        """Spark架構(gòu)概覽"""
        architecture = {
            "驅(qū)動(dòng)節(jié)點(diǎn)(Driver)": "執(zhí)行main方法，創(chuàng)建SparkContext",
            "集群管理器(Cluster Manager)": "資源分配和調(diào)度",
            "工作節(jié)點(diǎn)(Worker Node)": "執(zhí)行具體計(jì)算任務(wù)", 
            "執(zhí)行器(Executor)": "在工作節(jié)點(diǎn)上運(yùn)行任務(wù)"
        }
        return architecture

# PySpark優(yōu)勢(shì)分析示例
intro = PySparkIntroduction()
print("=== PySpark核心優(yōu)勢(shì) ===")
for category, items in intro.advantages.items():
    print(f"\n{category}:")
    for item in items:
        print(f"  ? {item}")

performance_df = intro.demonstrate_performance_comparison()
print("\n=== 性能對(duì)比分析 ===")
print(performance_df.to_string(index=False))

class PySparkEnvironment:
    """PySpark環(huán)境配置管理"""
    
    def __init__(self):
        self.spark = None
        self.config = {
            "spark.sql.adaptive.enabled": "true",
            "spark.sql.adaptive.coalescePartitions.enabled": "true",
            "spark.sql.adaptive.skew.enabled": "true",
            "spark.serializer": "org.apache.spark.serializer.KryoSerializer",
            "spark.memory.fraction": "0.8",
            "spark.memory.storageFraction": "0.3"
        }
    
    def create_spark_session(self, app_name="PySparkApp", master="local[*]", **kwargs):
        """創(chuàng)建Spark會(huì)話(huà)"""
        try:
            builder = SparkSession.builder \
                .appName(app_name) \
                .master(master)
            
            # 添加配置
            for key, value in self.config.items():
                builder = builder.config(key, value)
            
            # 添加額外配置
            for key, value in kwargs.items():
                builder = builder.config(key, value)
            
            self.spark = builder.getOrCreate()
            
            # 顯示配置信息
            print("? Spark會(huì)話(huà)創(chuàng)建成功")
            print(f"應(yīng)用名稱(chēng): {app_name}")
            print(f"運(yùn)行模式: {master}")
            print(f"Spark版本: {self.spark.version}")
            print(f"可用執(zhí)行器內(nèi)存: {self.spark.sparkContext.getConf().get('spark.executor.memory')}")
            
            return self.spark
            
        except Exception as e:
            print(f"? Spark會(huì)話(huà)創(chuàng)建失敗: {e}")
            return None
    
    def optimize_spark_config(self, data_size_gb):
        """根據(jù)數(shù)據(jù)大小優(yōu)化配置"""
        config_updates = {}
        
        if data_size_gb < 10:
            config_updates.update({
                "spark.sql.shuffle.partitions": "200",
                "spark.default.parallelism": "200"
            })
        elif data_size_gb < 100:
            config_updates.update({
                "spark.sql.shuffle.partitions": "1000", 
                "spark.default.parallelism": "1000"
            })
        else:
            config_updates.update({
                "spark.sql.shuffle.partitions": "2000",
                "spark.default.parallelism": "2000"
            })
        
        return config_updates
    
    def stop_spark_session(self):
        """停止Spark會(huì)話(huà)"""
        if self.spark:
            self.spark.stop()
            print("? Spark會(huì)話(huà)已停止")

# 環(huán)境配置演示
env = PySparkEnvironment()
spark = env.create_spark_session(
    app_name="BigDataProcessing",
    master="local[4]",
    spark_executor_memory="2g",
    spark_driver_memory="1g"
)

# 優(yōu)化配置示例
optimized_config = env.optimize_spark_config(50)
print("\n=== 優(yōu)化配置建議 ===")
for key, value in optimized_config.items():
    print(f"{key}: {value}")

class SparkCoreConcepts:
    """Spark核心概念解析"""
    
    def __init__(self, spark):
        self.spark = spark
        self.sc = spark.sparkContext
    
    def demonstrate_rdd_operations(self):
        """演示RDD基本操作"""
        print("=== RDD轉(zhuǎn)換與行動(dòng)操作演示 ===")
        
        # 創(chuàng)建RDD
        data = list(range(1, 101))
        rdd = self.sc.parallelize(data, 4)  # 4個(gè)分區(qū)
        
        print(f"RDD分區(qū)數(shù): {rdd.getNumPartitions()}")
        print(f"數(shù)據(jù)總量: {rdd.count()}")
        
        # 轉(zhuǎn)換操作 - 懶加載
        transformed_rdd = rdd \
            .filter(lambda x: x % 2 == 0) \
            .map(lambda x: x * x) \
            .map(lambda x: (x % 10, x))
        
        # 行動(dòng)操作 - 觸發(fā)計(jì)算
        result = transformed_rdd.reduceByKey(lambda a, b: a + b).collect()
        
        print("按最后一位數(shù)字分組求和結(jié)果:")
        for key, value in sorted(result):
            print(f"  數(shù)字結(jié)尾{key}: {value}")
        
        return transformed_rdd
    
    def demonstrate_lazy_evaluation(self):
        """演示懶加載機(jī)制"""
        print("\n=== 懶加載機(jī)制演示 ===")
        
        # 創(chuàng)建RDD
        rdd = self.sc.parallelize(range(1, 11))
        
        print("定義轉(zhuǎn)換操作...")
        # 轉(zhuǎn)換操作不會(huì)立即執(zhí)行
        mapped_rdd = rdd.map(lambda x: x * 2)
        filtered_rdd = mapped_rdd.filter(lambda x: x > 10)
        
        print("轉(zhuǎn)換操作定義完成，尚未執(zhí)行計(jì)算")
        print("觸發(fā)行動(dòng)操作...")
        
        # 行動(dòng)操作觸發(fā)計(jì)算
        result = filtered_rdd.collect()
        print(f"計(jì)算結(jié)果: {result}")
    
    def demonstrate_dataframe_creation(self):
        """演示DataFrame創(chuàng)建"""
        print("\n=== DataFrame創(chuàng)建演示 ===")
        
        # 方法1: 從Pandas DataFrame創(chuàng)建
        pandas_df = pd.DataFrame({
            'id': range(1, 6),
            'name': ['Alice', 'Bob', 'Charlie', 'David', 'Eve'],
            'age': [25, 30, 35, 28, 32],
            'salary': [50000, 60000, 70000, 55000, 65000]
        })
        
        spark_df1 = self.spark.createDataFrame(pandas_df)
        print("從Pandas創(chuàng)建DataFrame:")
        spark_df1.show()
        
        # 方法2: 通過(guò)Schema創(chuàng)建
        schema = StructType([
            StructField("product_id", IntegerType(), True),
            StructField("product_name", StringType(), True),
            StructField("price", DoubleType(), True),
            StructField("category", StringType(), True)
        ])
        
        data = [
            (1, "Laptop", 999.99, "Electronics"),
            (2, "Book", 29.99, "Education"), 
            (3, "Chair", 149.99, "Furniture")
        ]
        
        spark_df2 = self.spark.createDataFrame(data, schema)
        print("通過(guò)Schema創(chuàng)建DataFrame:")
        spark_df2.show()
        
        return spark_df1, spark_df2

# 核心概念演示
if spark:
    concepts = SparkCoreConcepts(spark)
    rdd_demo = concepts.demonstrate_rdd_operations()
    concepts.demonstrate_lazy_evaluation()
    df1, df2 = concepts.demonstrate_dataframe_creation()

class BigDataProcessor:
    """大規(guī)模數(shù)據(jù)處理器"""
    
    def __init__(self, spark):
        self.spark = spark
        self.processed_data = {}
    
    def read_multiple_data_sources(self, base_path):
        """讀取多種數(shù)據(jù)源"""
        print("=== 多數(shù)據(jù)源讀取 ===")
        
        try:
            # 讀取CSV文件
            csv_df = self.spark.read \
                .option("header", "true") \
                .option("inferSchema", "true") \
                .csv(f"{base_path}/*.csv")
            
            print(f"CSV數(shù)據(jù)記錄數(shù): {csv_df.count()}")
            
            # 讀取Parquet文件（列式存儲(chǔ)，更適合大數(shù)據(jù)）
            parquet_df = self.spark.read.parquet(f"{base_path}/*.parquet")
            print(f"Parquet數(shù)據(jù)記錄數(shù): {parquet_df.count()}")
            
            # 讀取JSON文件
            json_df = self.spark.read \
                .option("multiline", "true") \
                .json(f"{base_path}/*.json")
            print(f"JSON數(shù)據(jù)記錄數(shù): {json_df.count()}")
            
            return {
                "csv": csv_df,
                "parquet": parquet_df, 
                "json": json_df
            }
            
        except Exception as e:
            print(f"數(shù)據(jù)讀取失敗: {e}")
            return self.generate_sample_data()
    
    def generate_sample_data(self, num_records=100000):
        """生成模擬大數(shù)據(jù)集"""
        print("生成模擬大數(shù)據(jù)集...")
        
        # 用戶(hù)數(shù)據(jù)
        users_data = []
        for i in range(num_records):
            users_data.append((
                i + 1,  # user_id
                f"user_{i}@email.com",  # email
                np.random.choice(['北京', '上海', '廣州', '深圳', '杭州']),  # city
                np.random.randint(18, 65),  # age
                np.random.choice(['M', 'F']),  # gender
                np.random.normal(50000, 20000)  # income
            ))
        
        users_schema = StructType([
            StructField("user_id", IntegerType(), True),
            StructField("email", StringType(), True),
            StructField("city", StringType(), True),
            StructField("age", IntegerType(), True),
            StructField("gender", StringType(), True),
            StructField("income", DoubleType(), True)
        ])
        
        users_df = self.spark.createDataFrame(users_data, users_schema)
        
        # 交易數(shù)據(jù)
        transactions_data = []
        for i in range(num_records * 10):  # 10倍交易數(shù)據(jù)
            transactions_data.append((
                i + 1,  # transaction_id
                np.random.randint(1, num_records + 1),  # user_id
                np.random.choice(['Electronics', 'Clothing', 'Food', 'Books', 'Services']),  # category
                np.random.exponential(100),  # amount
                pd.Timestamp('2024-01-01') + pd.Timedelta(minutes=i),  # timestamp
                np.random.choice([True, False], p=[0.95, 0.05])  # is_successful
            ))
        
        transactions_schema = StructType([
            StructField("transaction_id", IntegerType(), True),
            StructField("user_id", IntegerType(), True),
            StructField("category", StringType(), True),
            StructField("amount", DoubleType(), True),
            StructField("timestamp", TimestampType(), True),
            StructField("is_successful", BooleanType(), True)
        ])
        
        transactions_df = self.spark.createDataFrame(transactions_data, transactions_schema)
        
        print(f"生成用戶(hù)數(shù)據(jù): {users_df.count():,} 條記錄")
        print(f"生成交易數(shù)據(jù): {transactions_df.count():,} 條記錄")
        
        return {
            "users": users_df,
            "transactions": transactions_df
        }
    
    def comprehensive_data_cleaning(self, df_dict):
        """綜合數(shù)據(jù)清洗"""
        print("\n=== 數(shù)據(jù)清洗流程 ===")
        
        cleaned_data = {}
        
        # 用戶(hù)數(shù)據(jù)清洗
        users_df = df_dict["users"]
        print(f"原始用戶(hù)數(shù)據(jù): {users_df.count():,} 條記錄")
        
        # 處理缺失值
        users_cleaned = users_df \
            .filter(col("user_id").isNotNull()) \
            .filter(col("email").isNotNull()) \
            .fillna({
                "age": users_df.select(mean("age")).first()[0],
                "income": users_df.select(mean("income")).first()[0]
            })
        
        # 處理異常值
        users_cleaned = users_cleaned \
            .filter((col("age") >= 18) & (col("age") <= 100)) \
            .filter(col("income") >= 0)
        
        print(f"清洗后用戶(hù)數(shù)據(jù): {users_cleaned.count():,} 條記錄")
        cleaned_data["users"] = users_cleaned
        
        # 交易數(shù)據(jù)清洗
        transactions_df = df_dict["transactions"]
        print(f"原始交易數(shù)據(jù): {transactions_df.count():,} 條記錄")
        
        transactions_cleaned = transactions_df \
            .filter(col("transaction_id").isNotNull()) \
            .filter(col("user_id").isNotNull()) \
            .filter(col("amount") > 0) \
            .filter(col("timestamp") >= '2024-01-01')
        
        print(f"清洗后交易數(shù)據(jù): {transactions_cleaned.count():,} 條記錄")
        cleaned_data["transactions"] = transactions_cleaned
        
        # 數(shù)據(jù)質(zhì)量報(bào)告
        self.generate_data_quality_report(cleaned_data)
        
        return cleaned_data
    
    def generate_data_quality_report(self, data_dict):
        """生成數(shù)據(jù)質(zhì)量報(bào)告"""
        print("\n=== 數(shù)據(jù)質(zhì)量報(bào)告 ===")
        
        for name, df in data_dict.items():
            total_count = df.count()
            
            # 計(jì)算各列的缺失值比例
            missing_stats = []
            for col_name in df.columns:
                missing_count = df.filter(col(col_name).isNull()).count()
                missing_ratio = missing_count / total_count if total_count > 0 else 0
                missing_stats.append((col_name, missing_ratio))
            
            print(f"\n{name} 數(shù)據(jù)質(zhì)量:")
            print(f"總記錄數(shù): {total_count:,}")
            print("各列缺失值比例:")
            for col_name, ratio in missing_stats:
                print(f"  {col_name}: {ratio:.3%}")

# 大數(shù)據(jù)處理演示
if spark:
    processor = BigDataProcessor(spark)
    
    # 生成模擬數(shù)據(jù)（在實(shí)際應(yīng)用中替換為真實(shí)數(shù)據(jù)路徑）
    raw_data = processor.generate_sample_data(50000)
    
    # 數(shù)據(jù)清洗
    cleaned_data = processor.comprehensive_data_cleaning(raw_data)

class AdvancedDataAnalyzer:
    """高級(jí)數(shù)據(jù)分析器"""
    
    def __init__(self, spark):
        self.spark = spark
    
    def perform_complex_aggregations(self, users_df, transactions_df):
        """執(zhí)行復(fù)雜聚合分析"""
        print("=== 復(fù)雜聚合分析 ===")
        
        # 1. 用戶(hù)行為分析
        user_behavior = transactions_df \
            .groupBy("user_id") \
            .agg(
                count("transaction_id").alias("transaction_count"),
                sum("amount").alias("total_spent"),
                avg("amount").alias("avg_transaction_amount"),
                max("timestamp").alias("last_transaction_date"),
                countDistinct("category").alias("unique_categories")
            ) \
            .join(users_df, "user_id", "inner")
        
        print("用戶(hù)行為分析:")
        user_behavior.select("user_id", "transaction_count", "total_spent", "city").show(10)
        
        # 2. 城市級(jí)銷(xiāo)售分析
        city_sales = transactions_df \
            .join(users_df, "user_id") \
            .groupBy("city") \
            .agg(
                count("transaction_id").alias("total_transactions"),
                sum("amount").alias("total_revenue"),
                avg("amount").alias("avg_transaction_value"),
                countDistinct("user_id").alias("unique_customers")
            ) \
            .withColumn("revenue_per_customer", col("total_revenue") / col("unique_customers")) \
            .orderBy(col("total_revenue").desc())
        
        print("\n城市銷(xiāo)售分析:")
        city_sales.show()
        
        # 3. 時(shí)間序列分析
        from pyspark.sql.functions import date_format
        
        daily_sales = transactions_df \
            .withColumn("date", date_format("timestamp", "yyyy-MM-dd")) \
            .groupBy("date") \
            .agg(
                count("transaction_id").alias("daily_transactions"),
                sum("amount").alias("daily_revenue"),
                avg("amount").alias("avg_daily_transaction")
            ) \
            .orderBy("date")
        
        print("\n每日銷(xiāo)售趨勢(shì):")
        daily_sales.show(10)
        
        return {
            "user_behavior": user_behavior,
            "city_sales": city_sales, 
            "daily_sales": daily_sales
        }
    
    def window_function_analysis(self, transactions_df):
        """窗口函數(shù)分析"""
        print("\n=== 窗口函數(shù)分析 ===")
        
        from pyspark.sql.window import Window
        
        # 定義窗口規(guī)范
        user_window = Window \
            .partitionBy("user_id") \
            .orderBy("timestamp") \
            .rowsBetween(Window.unboundedPreceding, Window.currentRow)
        
        # 使用窗口函數(shù)計(jì)算累計(jì)值
        user_cumulative = transactions_df \
            .withColumn("cumulative_spent", sum("amount").over(user_window)) \
            .withColumn("transaction_rank", row_number().over(user_window)) \
            .withColumn("prev_amount", lag("amount", 1).over(user_window))
        
        print("用戶(hù)累計(jì)消費(fèi)分析:")
        user_cumulative.filter(col("user_id") <= 5).select(
            "user_id", "timestamp", "amount", "cumulative_spent", "transaction_rank"
        ).show(20)
        
        return user_cumulative
    
    def advanced_analytics_with_pandas_udf(self, users_df, transactions_df):
        """使用Pandas UDF進(jìn)行高級(jí)分析"""
        print("\n=== 使用Pandas UDF進(jìn)行高級(jí)分析 ===")
        
        from pyspark.sql.functions import pandas_udf
        from pyspark.sql.types import DoubleType
        
        # 定義Pandas UDF計(jì)算用戶(hù)價(jià)值評(píng)分
        @pandas_udf(DoubleType())
        def calculate_customer_value(transaction_counts, total_spent, unique_categories):
            """計(jì)算客戶(hù)價(jià)值評(píng)分"""
            # 使用RFM-like評(píng)分機(jī)制
            frequency_score = np.log1p(transaction_counts) / np.log1p(transaction_counts.max())
            monetary_score = total_spent / total_spent.max()
            variety_score = unique_categories / unique_categories.max()
            
            # 綜合評(píng)分（加權(quán)平均）
            overall_score = 0.4 * frequency_score + 0.4 * monetary_score + 0.2 * variety_score
            return overall_score
        
        # 準(zhǔn)備數(shù)據(jù)
        user_metrics = transactions_df \
            .groupBy("user_id") \
            .agg(
                count("transaction_id").alias("transaction_count"),
                sum("amount").alias("total_spent"),
                countDistinct("category").alias("unique_categories")
            )
        
        # 應(yīng)用Pandas UDF
        user_value_analysis = user_metrics \
            .withColumn("customer_value_score", 
                       calculate_customer_value("transaction_count", "total_spent", "unique_categories")) \
            .join(users_df, "user_id") \
            .orderBy(col("customer_value_score").desc())
        
        print("客戶(hù)價(jià)值分析:")
        user_value_analysis.select("user_id", "city", "transaction_count", 
                                 "total_spent", "customer_value_score").show(10)
        
        return user_value_analysis

# 高級(jí)分析演示
if spark:
    analyzer = AdvancedDataAnalyzer(spark)
    
    # 執(zhí)行復(fù)雜分析
    analysis_results = analyzer.perform_complex_aggregations(
        cleaned_data["users"], cleaned_data["transactions"]
    )
    
    # 窗口函數(shù)分析
    window_analysis = analyzer.window_function_analysis(cleaned_data["transactions"])
    
    # Pandas UDF分析
    value_analysis = analyzer.advanced_analytics_with_pandas_udf(
        cleaned_data["users"], cleaned_data["transactions"]
    )

class PerformanceOptimizer:
    """PySpark性能優(yōu)化器"""
    
    def __init__(self, spark):
        self.spark = spark
    
    def analyze_query_plan(self, df, description):
        """分析查詢(xún)執(zhí)行計(jì)劃"""
        print(f"\n=== {description} 執(zhí)行計(jì)劃分析 ===")
        
        # 顯示邏輯計(jì)劃
        print("邏輯執(zhí)行計(jì)劃:")
        print(df._jdf.queryExecution().logical().toString())
        
        # 顯示物理計(jì)劃  
        print("\n物理執(zhí)行計(jì)劃:")
        print(df._jdf.queryExecution().executedPlan().toString())
        
        # 顯示優(yōu)化計(jì)劃
        print("\n優(yōu)化后的執(zhí)行計(jì)劃:")
        print(df._jdf.queryExecution().optimizedPlan().toString())
    
    def demonstrate_caching_strategies(self, df):
        """演示緩存策略"""
        print("\n=== 緩存策略演示 ===")
        
        import time
        
        # 不緩存的情況
        start_time = time.time()
        result1 = df.groupBy("city").agg(sum("amount").alias("total")).collect()
        time1 = time.time() - start_time
        
        # 緩存后的情況
        df.cache()
        df.count()  # 觸發(fā)緩存
        
        start_time = time.time()
        result2 = df.groupBy("city").agg(sum("amount").alias("total")).collect()
        time2 = time.time() - start_time
        
        print(f"未緩存執(zhí)行時(shí)間: {time1:.4f}秒")
        print(f"緩存后執(zhí)行時(shí)間: {time2:.4f}秒")
        print(f"性能提升: {time1/time2:.2f}x")
        
        # 清理緩存
        df.unpersist()
    
    def partition_optimization(self, df, partition_col):
        """分區(qū)優(yōu)化"""
        print(f"\n=== 分區(qū)優(yōu)化: {partition_col} ===")
        
        # 檢查當(dāng)前分區(qū)數(shù)
        initial_partitions = df.rdd.getNumPartitions()
        print(f"初始分區(qū)數(shù): {initial_partitions}")
        
        # 重新分區(qū)
        optimized_df = df.repartition(200, partition_col)
        optimized_partitions = optimized_df.rdd.getNumPartitions()
        print(f"優(yōu)化后分區(qū)數(shù): {optimized_partitions}")
        
        # 顯示分區(qū)統(tǒng)計(jì)
        partition_stats = optimized_df \
            .groupBy(spark_partition_id().alias("partition_id")) \
            .count() \
            .orderBy("partition_id")
        
        print("分區(qū)數(shù)據(jù)分布:")
        partition_stats.show(10)
        
        return optimized_df
    
    def broadcast_join_optimization(self, large_df, small_df):
        """廣播連接優(yōu)化"""
        print("\n=== 廣播連接優(yōu)化 ===")
        
        from pyspark.sql.functions import broadcast
        
        # 標(biāo)準(zhǔn)連接
        start_time = time.time()
        standard_join = large_df.join(small_df, "user_id")
        standard_count = standard_join.count()
        standard_time = time.time() - start_time
        
        # 廣播連接
        start_time = time.time()
        broadcast_join = large_df.join(broadcast(small_df), "user_id")
        broadcast_count = broadcast_join.count()
        broadcast_time = time.time() - start_time
        
        print(f"標(biāo)準(zhǔn)連接 - 記錄數(shù): {standard_count:,}, 時(shí)間: {standard_time:.2f}秒")
        print(f"廣播連接 - 記錄數(shù): {broadcast_count:,}, 時(shí)間: {broadcast_time:.2f}秒")
        print(f"性能提升: {standard_time/broadcast_time:.2f}x")
        
        return broadcast_join

# 性能優(yōu)化演示
if spark:
    optimizer = PerformanceOptimizer(spark)
    
    # 分析執(zhí)行計(jì)劃
    sample_df = cleaned_data["transactions"].filter(col("amount") > 50)
    optimizer.analyze_query_plan(sample_df, "過(guò)濾交易數(shù)據(jù)")
    
    # 緩存策略演示
    optimizer.demonstrate_caching_strategies(cleaned_data["transactions"])
    
    # 分區(qū)優(yōu)化
    partitioned_df = optimizer.partition_optimization(
        cleaned_data["transactions"], "category"
    )

#!/usr/bin/env python3
"""
ecommerce_user_analysis.py
電商用戶(hù)行為分析系統(tǒng) - 完整PySpark實(shí)現(xiàn)
"""

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *
from pyspark.sql.window import Window
from pyspark.ml.feature import VectorAssembler, StandardScaler
from pyspark.ml.clustering import KMeans
from pyspark.ml.evaluation import ClusteringEvaluator
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import time

class EcommerceUserAnalysis:
    """電商用戶(hù)行為分析系統(tǒng)"""
    
    def __init__(self):
        self.spark = self.initialize_spark_session()
        self.analysis_results = {}
    
    def initialize_spark_session(self):
        """初始化Spark會(huì)話(huà)"""
        spark = SparkSession.builder \
            .appName("EcommerceUserAnalysis") \
            .master("local[4]") \
            .config("spark.sql.adaptive.enabled", "true") \
            .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \
            .config("spark.executor.memory", "2g") \
            .config("spark.driver.memory", "1g") \
            .config("spark.sql.shuffle.partitions", "200") \
            .getOrCreate()
        
        print("? Spark會(huì)話(huà)初始化完成")
        return spark
    
    def generate_ecommerce_data(self, num_users=100000, num_transactions=1000000):
        """生成電商模擬數(shù)據(jù)"""
        print("生成電商模擬數(shù)據(jù)...")
        
        # 用戶(hù)數(shù)據(jù)
        users_data = []
        cities = ['北京', '上海', '廣州', '深圳', '杭州', '成都', '武漢', '西安', '南京', '重慶']
        
        for i in range(num_users):
            users_data.append((
                i + 1,  # user_id
                f"user_{i}@example.com",  # email
                np.random.choice(cities),  # city
                np.random.randint(18, 70),  # age
                np.random.choice(['M', 'F']),  # gender
                np.random.normal(50000, 20000),  # annual_income
                datetime.now() - timedelta(days=np.random.randint(1, 365*3))  # registration_date
            ))
        
        users_schema = StructType([
            StructField("user_id", IntegerType(), True),
            StructField("email", StringType(), True),
            StructField("city", StringType(), True),
            StructField("age", IntegerType(), True),
            StructField("gender", StringType(), True),
            StructField("annual_income", DoubleType(), True),
            StructField("registration_date", TimestampType(), True)
        ])
        
        users_df = self.spark.createDataFrame(users_data, users_schema)
        
        # 交易數(shù)據(jù)
        transactions_data = []
        categories = ['Electronics', 'Clothing', 'Food', 'Books', 'Home', 'Beauty', 'Sports', 'Toys']
        
        for i in range(num_transactions):
            transactions_data.append((
                i + 1,  # transaction_id
                np.random.randint(1, num_users + 1),  # user_id
                np.random.choice(categories),  # category
                np.random.exponential(150),  # amount
                datetime.now() - timedelta(hours=np.random.randint(1, 24*30)),  # timestamp
                np.random.choice([True, False], p=[0.97, 0.03]),  # is_successful
                np.random.choice([1, 2, 3, 4, 5], p=[0.1, 0.15, 0.5, 0.2, 0.05])  # rating
            ))
        
        transactions_schema = StructType([
            StructField("transaction_id", IntegerType(), True),
            StructField("user_id", IntegerType(), True),
            StructField("category", StringType(), True),
            StructField("amount", DoubleType(), True),
            StructField("timestamp", TimestampType(), True),
            StructField("is_successful", BooleanType(), True),
            StructField("rating", IntegerType(), True)
        ])
        
        transactions_df = self.spark.createDataFrame(transactions_data, transactions_schema)
        
        print(f"? 生成用戶(hù)數(shù)據(jù): {users_df.count():,} 條")
        print(f"? 生成交易數(shù)據(jù): {transactions_df.count():,} 條")
        
        return users_df, transactions_df
    
    def comprehensive_user_analysis(self, users_df, transactions_df):
        """綜合用戶(hù)行為分析"""
        print("\n" + "="*50)
        print("開(kāi)始綜合用戶(hù)行為分析")
        print("="*50)
        
        # 1. 用戶(hù)基本行為分析
        user_behavior = self.analyze_user_behavior(users_df, transactions_df)
        
        # 2. RFM分析
        rfm_analysis = self.perform_rfm_analysis(transactions_df)
        
        # 3. 用戶(hù)聚類(lèi)分析
        user_clusters = self.perform_user_clustering(user_behavior)
        
        # 4. 時(shí)間序列分析
        time_analysis = self.analyze_temporal_patterns(transactions_df)
        
        # 5. 生成業(yè)務(wù)洞察
        business_insights = self.generate_business_insights(
            user_behavior, rfm_analysis, user_clusters, time_analysis
        )
        
        self.analysis_results = {
            "user_behavior": user_behavior,
            "rfm_analysis": rfm_analysis,
            "user_clusters": user_clusters,
            "time_analysis": time_analysis,
            "business_insights": business_insights
        }
        
        return self.analysis_results
    
    def analyze_user_behavior(self, users_df, transactions_df):
        """用戶(hù)行為分析"""
        print("進(jìn)行用戶(hù)行為分析...")
        
        user_behavior = transactions_df \
            .filter(col("is_successful") == True) \
            .groupBy("user_id") \
            .agg(
                count("transaction_id").alias("transaction_count"),
                sum("amount").alias("total_spent"),
                avg("amount").alias("avg_transaction_value"),
                countDistinct("category").alias("unique_categories"),
                avg("rating").alias("avg_rating"),
                max("timestamp").alias("last_transaction_date"),
                min("timestamp").alias("first_transaction_date")
            ) \
            .withColumn("customer_lifetime_days", 
                       datediff(col("last_transaction_date"), col("first_transaction_date"))) \
            .withColumn("avg_days_between_transactions", 
                       col("customer_lifetime_days") / col("transaction_count")) \
            .join(users_df, "user_id", "inner")
        
        print(f"? 用戶(hù)行為分析完成，分析 {user_behavior.count():,} 名用戶(hù)")
        return user_behavior
    
    def perform_rfm_analysis(self, transactions_df):
        """RFM分析（最近購(gòu)買(mǎi)、購(gòu)買(mǎi)頻率、購(gòu)買(mǎi)金額）"""
        print("進(jìn)行RFM分析...")
        
        # 計(jì)算基準(zhǔn)日期（最近30天）
        max_date = transactions_df.select(max("timestamp")).first()[0]
        baseline_date = max_date - timedelta(days=30)
        
        # RFM計(jì)算
        rfm_data = transactions_df \
            .filter(col("is_successful") == True) \
            .filter(col("timestamp") >= baseline_date) \
            .groupBy("user_id") \
            .agg(
                datediff(lit(max_date), max("timestamp")).alias("recency"),  # 最近購(gòu)買(mǎi)
                count("transaction_id").alias("frequency"),  # 購(gòu)買(mǎi)頻率
                sum("amount").alias("monetary")  # 購(gòu)買(mǎi)金額
            ) \
            .filter(col("frequency") > 0)  # 只分析有購(gòu)買(mǎi)行為的用戶(hù)
        
        # RFM評(píng)分（5分制）
        recency_window = Window.orderBy("recency")
        frequency_window = Window.orderBy(col("frequency").desc())
        monetary_window = Window.orderBy(col("monetary").desc())
        
        rfm_scored = rfm_data \
            .withColumn("r_score", ntile(5).over(recency_window)) \
            .withColumn("f_score", ntile(5).over(frequency_window)) \
            .withColumn("m_score", ntile(5).over(monetary_window)) \
            .withColumn("rfm_score", col("r_score") + col("f_score") + col("m_score")) \
            .withColumn("rfm_segment",
                       when(col("rfm_score") >= 12, "冠軍客戶(hù)")
                       .when(col("rfm_score") >= 9, "忠實(shí)客戶(hù)") 
                       .when(col("rfm_score") >= 6, "潛力客戶(hù)")
                       .when(col("rfm_score") >= 3, "新客戶(hù)")
                       .otherwise("流失風(fēng)險(xiǎn)客戶(hù)"))
        
        print(f"? RFM分析完成，分析 {rfm_scored.count():,} 名用戶(hù)")
        return rfm_scored
    
    def perform_user_clustering(self, user_behavior):
        """用戶(hù)聚類(lèi)分析"""
        print("進(jìn)行用戶(hù)聚類(lèi)分析...")
        
        # 準(zhǔn)備特征
        feature_cols = ["transaction_count", "total_spent", "unique_categories", "avg_rating"]
        
        # 處理缺失值
        clustering_data = user_behavior \
            .filter(col("transaction_count") > 1) \
            .fillna(0, subset=feature_cols)
        
        # 特征向量化
        assembler = VectorAssembler(inputCols=feature_cols, outputCol="features")
        feature_vector = assembler.transform(clustering_data)
        
        # 特征標(biāo)準(zhǔn)化
        scaler = StandardScaler(inputCol="features", outputCol="scaled_features")
        scaler_model = scaler.fit(feature_vector)
        scaled_data = scaler_model.transform(feature_vector)
        
        # K-means聚類(lèi)
        kmeans = KMeans(featuresCol="scaled_features", k=4, seed=42)
        kmeans_model = kmeans.fit(scaled_data)
        clustered_data = kmeans_model.transform(scaled_data)
        
        # 評(píng)估聚類(lèi)效果
        evaluator = ClusteringEvaluator(featuresCol="scaled_features")
        silhouette_score = evaluator.evaluate(clustered_data)
        
        print(f"? 用戶(hù)聚類(lèi)完成，輪廓系數(shù): {silhouette_score:.3f}")
        
        # 分析聚類(lèi)特征
        cluster_profiles = clustered_data \
            .groupBy("prediction") \
            .agg(
                count("user_id").alias("cluster_size"),
                avg("transaction_count").alias("avg_transactions"),
                avg("total_spent").alias("avg_spent"),
                avg("unique_categories").alias("avg_categories"),
                avg("avg_rating").alias("avg_rating")
            ) \
            .orderBy("prediction")
        
        print("聚類(lèi)特征分析:")
        cluster_profiles.show()
        
        return {
            "clustered_data": clustered_data,
            "cluster_profiles": cluster_profiles,
            "silhouette_score": silhouette_score
        }
    
    def analyze_temporal_patterns(self, transactions_df):
        """時(shí)間序列模式分析"""
        print("進(jìn)行時(shí)間序列分析...")
        
        # 按小時(shí)分析購(gòu)買(mǎi)模式
        hourly_patterns = transactions_df \
            .filter(col("is_successful") == True) \
            .withColumn("hour", hour("timestamp")) \
            .groupBy("hour") \
            .agg(
                count("transaction_id").alias("transaction_count"),
                avg("amount").alias("avg_amount"),
                countDistinct("user_id").alias("unique_users")
            ) \
            .orderBy("hour")
        
        # 按星期分析購(gòu)買(mǎi)模式
        daily_patterns = transactions_df \
            .filter(col("is_successful") == True) \
            .withColumn("day_of_week", date_format("timestamp", "E")) \
            .groupBy("day_of_week") \
            .agg(
                count("transaction_id").alias("transaction_count"),
                avg("amount").alias("avg_amount")
            ) \
            .orderBy("day_of_week")
        
        # 月度趨勢(shì)分析
        monthly_trends = transactions_df \
            .filter(col("is_successful") == True) \
            .withColumn("month", date_format("timestamp", "yyyy-MM")) \
            .groupBy("month") \
            .agg(
                count("transaction_id").alias("transaction_count"),
                sum("amount").alias("total_revenue"),
                countDistinct("user_id").alias("unique_customers")
            ) \
            .orderBy("month")
        
        print("? 時(shí)間序列分析完成")
        
        return {
            "hourly_patterns": hourly_patterns,
            "daily_patterns": daily_patterns, 
            "monthly_trends": monthly_trends
        }
    
    def generate_business_insights(self, user_behavior, rfm_analysis, user_clusters, time_analysis):
        """生成業(yè)務(wù)洞察"""
        print("生成業(yè)務(wù)洞察...")
        
        insights = {}
        
        # 1. 關(guān)鍵指標(biāo)
        total_users = user_behavior.count()
        total_revenue = user_behavior.agg(sum("total_spent")).first()[0]
        avg_transaction_value = user_behavior.agg(avg("avg_transaction_value")).first()[0]
        
        insights["key_metrics"] = {
            "total_users": total_users,
            "total_revenue": total_revenue,
            "avg_transaction_value": avg_transaction_value,
            "avg_customer_rating": user_behavior.agg(avg("avg_rating")).first()[0]
        }
        
        # 2. RFM細(xì)分統(tǒng)計(jì)
        rfm_segment_stats = rfm_analysis \
            .groupBy("rfm_segment") \
            .agg(count("user_id").alias("user_count")) \
            .orderBy(col("user_count").desc())
        
        insights["rfm_segments"] = {
            row["rfm_segment"]: row["user_count"] 
            for row in rfm_segment_stats.collect()
        }
        
        # 3. 聚類(lèi)分析洞察
        cluster_insights = user_clusters["cluster_profiles"].collect()
        insights["cluster_analysis"] = [
            {
                "cluster_id": row["prediction"],
                "size": row["cluster_size"],
                "avg_transactions": row["avg_transactions"],
                "avg_spent": row["avg_spent"]
            }
            for row in cluster_insights
        ]
        
        # 4. 時(shí)間模式洞察
        peak_hour = time_analysis["hourly_patterns"] \
            .orderBy(col("transaction_count").desc()) \
            .first()
        
        insights["temporal_insights"] = {
            "peak_hour": peak_hour["hour"],
            "peak_hour_transactions": peak_hour["transaction_count"],
            "busiest_day": time_analysis["daily_patterns"]
                .orderBy(col("transaction_count").desc())
                .first()["day_of_week"]
        }
        
        print("? 業(yè)務(wù)洞察生成完成")
        return insights
    
    def generate_comprehensive_report(self):
        """生成綜合分析報(bào)告"""
        if not self.analysis_results:
            print("請(qǐng)先執(zhí)行分析")
            return
        
        insights = self.analysis_results["business_insights"]
        
        print("\n" + "="*60)
        print("電商用戶(hù)行為分析報(bào)告")
        print("="*60)
        
        # 關(guān)鍵指標(biāo)
        print("\n?? 關(guān)鍵業(yè)務(wù)指標(biāo):")
        metrics = insights["key_metrics"]
        print(f"  ? 總用戶(hù)數(shù): {metrics['total_users']:,}")
        print(f"  ? 總營(yíng)收: ￥{metrics['total_revenue']:,.2f}")
        print(f"  ? 平均交易價(jià)值: ￥{metrics['avg_transaction_value']:.2f}")
        print(f"  ? 平均客戶(hù)評(píng)分: {metrics['avg_customer_rating']:.2f}/5")
        
        # RFM細(xì)分
        print("\n?? RFM客戶(hù)細(xì)分:")
        for segment, count in insights["rfm_segments"].items():
            percentage = (count / metrics['total_users']) * 100
            print(f"  ? {segment}: {count:,} 人 ({percentage:.1f}%)")
        
        # 聚類(lèi)分析
        print("\n?? 用戶(hù)聚類(lèi)分析:")
        for cluster in insights["cluster_analysis"]:
            print(f"  ? 聚類(lèi){cluster['cluster_id']}: {cluster['size']:,} 用戶(hù)")
            print(f"    平均交易數(shù): {cluster['avg_transactions']:.1f}")
            print(f"    平均消費(fèi): ￥{cluster['avg_spent']:,.2f}")
        
        # 時(shí)間洞察
        print("\n? 時(shí)間模式洞察:")
        temporal = insights["temporal_insights"]
        print(f"  ? 高峰時(shí)段: {temporal['peak_hour']}:00 ({temporal['peak_hour_transactions']} 筆交易)")
        print(f"  ? 最繁忙日期: {temporal['busiest_day']}")
        
        # 性能指標(biāo)
        if "silhouette_score" in self.analysis_results["user_clusters"]:
            score = self.analysis_results["user_clusters"]["silhouette_score"]
            print(f"\n?? 聚類(lèi)質(zhì)量: {score:.3f} (輪廓系數(shù))")
    
    def save_analysis_results(self, output_path):
        """保存分析結(jié)果"""
        print(f"\n保存分析結(jié)果到: {output_path}")
        
        try:
            # 保存用戶(hù)行為數(shù)據(jù)
            self.analysis_results["user_behavior"] \
                .write \
                .mode("overwrite") \
                .parquet(f"{output_path}/user_behavior")
            
            # 保存RFM分析結(jié)果
            self.analysis_results["rfm_analysis"] \
                .write \
                .mode("overwrite") \
                .parquet(f"{output_path}/rfm_analysis")
            
            # 保存聚類(lèi)結(jié)果
            self.analysis_results["user_clusters"]["clustered_data"] \
                .write \
                .mode("overwrite") \
                .parquet(f"{output_path}/user_clusters")
            
            print("? 分析結(jié)果保存完成")
            
        except Exception as e:
            print(f"? 保存失敗: {e}")
    
    def stop(self):
        """停止Spark會(huì)話(huà)"""
        self.spark.stop()
        print("? Spark會(huì)話(huà)已停止")

def main():
    """主函數(shù)"""
    print("啟動(dòng)電商用戶(hù)行為分析系統(tǒng)...")
    
    # 初始化分析系統(tǒng)
    analyzer = EcommerceUserAnalysis()
    
    try:
        # 1. 生成數(shù)據(jù)
        users_df, transactions_df = analyzer.generate_ecommerce_data(50000, 500000)
        
        # 2. 執(zhí)行分析
        analysis_results = analyzer.comprehensive_user_analysis(users_df, transactions_df)
        
        # 3. 生成報(bào)告
        analyzer.generate_comprehensive_report()
        
        # 4. 保存結(jié)果（在實(shí)際環(huán)境中取消注釋?zhuān)?
        # analyzer.save_analysis_results("hdfs://path/to/output")
        
        print("\n?? 分析完成!")
        
    except Exception as e:
        print(f"? 分析過(guò)程中出現(xiàn)錯(cuò)誤: {e}")
    
    finally:
        # 清理資源
        analyzer.stop()

if __name__ == "__main__":
    main()

class ProductionDeployment:
    """生產(chǎn)環(huán)境部署配置"""
    
    @staticmethod
    def get_cluster_configurations():
        """獲取集群配置模板"""
        configs = {
            "development": {
                "spark.master": "local[4]",
                "spark.executor.memory": "2g",
                "spark.driver.memory": "1g",
                "spark.sql.shuffle.partitions": "200"
            },
            "staging": {
                "spark.master": "spark://staging-cluster:7077",
                "spark.executor.memory": "8g", 
                "spark.driver.memory": "4g",
                "spark.executor.instances": "10",
                "spark.sql.shuffle.partitions": "1000"
            },
            "production": {
                "spark.master": "spark://prod-cluster:7077",
                "spark.executor.memory": "16g",
                "spark.driver.memory": "8g", 
                "spark.executor.instances": "50",
                "spark.sql.adaptive.enabled": "true",
                "spark.sql.adaptive.coalescePartitions.enabled": "true",
                "spark.sql.shuffle.partitions": "2000"
            }
        }
        return configs
    
    @staticmethod
    def create_production_session(app_name, environment="production"):
        """創(chuàng)建生產(chǎn)環(huán)境Spark會(huì)話(huà)"""
        configs = ProductionDeployment.get_cluster_configurations()
        config = configs.get(environment, configs["development"])
        
        builder = SparkSession.builder.appName(app_name)
        
        for key, value in config.items():
            builder = builder.config(key, value)
        
        return builder.getOrCreate()

# 生產(chǎn)配置示例
production_config = ProductionDeployment.get_cluster_configurations()["production"]
print("=== 生產(chǎn)環(huán)境配置 ===")
for key, value in production_config.items():
    print(f"{key}: {value}")

class OptimizationChecklist:
    """性能優(yōu)化檢查清單"""
    
    @staticmethod
    def get_checklist():
        """獲取優(yōu)化檢查清單"""
        return {
            "數(shù)據(jù)讀取": [
                "使用列式存儲(chǔ)格式(Parquet/ORC)",
                "合理設(shè)置分區(qū)數(shù)",
                "使用謂詞下推優(yōu)化"
            ],
            "數(shù)據(jù)處理": [
                "避免不必要的shuffle操作",
                "使用廣播連接小表",
                "合理使用緩存策略",
                "盡早過(guò)濾不需要的數(shù)據(jù)"
            ],
            "內(nèi)存管理": [
                "監(jiān)控Executor內(nèi)存使用",
                "合理設(shè)置序列化器",
                "避免數(shù)據(jù)傾斜",
                "使用堆外內(nèi)存"
            ],
            "執(zhí)行優(yōu)化": [
                "啟用自適應(yīng)查詢(xún)執(zhí)行",
                "合理設(shè)置并行度",
                "使用向量化UDF",
                "優(yōu)化數(shù)據(jù)本地性"
            ]
        }
    
    @staticmethod
    def validate_configuration(spark_conf):
        """驗(yàn)證配置合理性"""
        checks = {
            "adequate_memory": spark_conf.get("spark.executor.memory", "1g") >= "4g",
            "adaptive_enabled": spark_conf.get("spark.sql.adaptive.enabled", "false") == "true",
            "proper_parallelism": int(spark_conf.get("spark.sql.shuffle.partitions", "200")) >= 200,
            "kryo_serializer": spark_conf.get("spark.serializer", "").endswith("KryoSerializer")
        }
        
        return checks

# 優(yōu)化檢查清單
checklist = OptimizationChecklist()
print("=== PySpark性能優(yōu)化檢查清單 ===")
for category, items in checklist.get_checklist().items():
    print(f"\n{category}:")
    for item in items:
        print(f"  ? {item}")