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

# PySpark優(yōu)勢分析示例
intro = PySparkIntroduction()
print("=== PySpark核心優(yōu)勢 ===")
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=== 性能對比分析 ===")
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會話"""
        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會話創(chuàng)建成功")
            print(f"應用名稱: {app_name}")
            print(f"運行模式: {master}")
            print(f"Spark版本: {self.spark.version}")
            print(f"可用執(zhí)行器內存: {self.spark.sparkContext.getConf().get('spark.executor.memory')}")
            
            return self.spark
            
        except Exception as e:
            print(f"? Spark會話創(chuàng)建失敗: {e}")
            return None
    
    def optimize_spark_config(self, data_size_gb):
        """根據數據大小優(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會話"""
        if self.spark:
            self.spark.stop()
            print("? Spark會話已停止")

# 環(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轉換與行動操作演示 ===")
        
        # 創(chuàng)建RDD
        data = list(range(1, 101))
        rdd = self.sc.parallelize(data, 4)  # 4個分區(qū)
        
        print(f"RDD分區(qū)數: {rdd.getNumPartitions()}")
        print(f"數據總量: {rdd.count()}")
        
        # 轉換操作 - 懶加載
        transformed_rdd = rdd \
            .filter(lambda x: x % 2 == 0) \
            .map(lambda x: x * x) \
            .map(lambda x: (x % 10, x))
        
        # 行動操作 - 觸發(fā)計算
        result = transformed_rdd.reduceByKey(lambda a, b: a + b).collect()
        
        print("按最后一位數字分組求和結果:")
        for key, value in sorted(result):
            print(f"  數字結尾{key}: {value}")
        
        return transformed_rdd
    
    def demonstrate_lazy_evaluation(self):
        """演示懶加載機制"""
        print("\n=== 懶加載機制演示 ===")
        
        # 創(chuàng)建RDD
        rdd = self.sc.parallelize(range(1, 11))
        
        print("定義轉換操作...")
        # 轉換操作不會立即執(zhí)行
        mapped_rdd = rdd.map(lambda x: x * 2)
        filtered_rdd = mapped_rdd.filter(lambda x: x > 10)
        
        print("轉換操作定義完成，尚未執(zhí)行計算")
        print("觸發(fā)行動操作...")
        
        # 行動操作觸發(fā)計算
        result = filtered_rdd.collect()
        print(f"計算結果: {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: 通過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("通過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ī)模數據處理器"""
    
    def __init__(self, spark):
        self.spark = spark
        self.processed_data = {}
    
    def read_multiple_data_sources(self, base_path):
        """讀取多種數據源"""
        print("=== 多數據源讀取 ===")
        
        try:
            # 讀取CSV文件
            csv_df = self.spark.read \
                .option("header", "true") \
                .option("inferSchema", "true") \
                .csv(f"{base_path}/*.csv")
            
            print(f"CSV數據記錄數: {csv_df.count()}")
            
            # 讀取Parquet文件（列式存儲，更適合大數據）
            parquet_df = self.spark.read.parquet(f"{base_path}/*.parquet")
            print(f"Parquet數據記錄數: {parquet_df.count()}")
            
            # 讀取JSON文件
            json_df = self.spark.read \
                .option("multiline", "true") \
                .json(f"{base_path}/*.json")
            print(f"JSON數據記錄數: {json_df.count()}")
            
            return {
                "csv": csv_df,
                "parquet": parquet_df, 
                "json": json_df
            }
            
        except Exception as e:
            print(f"數據讀取失敗: {e}")
            return self.generate_sample_data()
    
    def generate_sample_data(self, num_records=100000):
        """生成模擬大數據集"""
        print("生成模擬大數據集...")
        
        # 用戶數據
        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)
        
        # 交易數據
        transactions_data = []
        for i in range(num_records * 10):  # 10倍交易數據
            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"生成用戶數據: {users_df.count():,} 條記錄")
        print(f"生成交易數據: {transactions_df.count():,} 條記錄")
        
        return {
            "users": users_df,
            "transactions": transactions_df
        }
    
    def comprehensive_data_cleaning(self, df_dict):
        """綜合數據清洗"""
        print("\n=== 數據清洗流程 ===")
        
        cleaned_data = {}
        
        # 用戶數據清洗
        users_df = df_dict["users"]
        print(f"原始用戶數據: {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"清洗后用戶數據: {users_cleaned.count():,} 條記錄")
        cleaned_data["users"] = users_cleaned
        
        # 交易數據清洗
        transactions_df = df_dict["transactions"]
        print(f"原始交易數據: {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"清洗后交易數據: {transactions_cleaned.count():,} 條記錄")
        cleaned_data["transactions"] = transactions_cleaned
        
        # 數據質量報告
        self.generate_data_quality_report(cleaned_data)
        
        return cleaned_data
    
    def generate_data_quality_report(self, data_dict):
        """生成數據質量報告"""
        print("\n=== 數據質量報告 ===")
        
        for name, df in data_dict.items():
            total_count = df.count()
            
            # 計算各列的缺失值比例
            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} 數據質量:")
            print(f"總記錄數: {total_count:,}")
            print("各列缺失值比例:")
            for col_name, ratio in missing_stats:
                print(f"  {col_name}: {ratio:.3%}")

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

class AdvancedDataAnalyzer:
    """高級數據分析器"""
    
    def __init__(self, spark):
        self.spark = spark
    
    def perform_complex_aggregations(self, users_df, transactions_df):
        """執(zhí)行復雜聚合分析"""
        print("=== 復雜聚合分析 ===")
        
        # 1. 用戶行為分析
        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("用戶行為分析:")
        user_behavior.select("user_id", "transaction_count", "total_spent", "city").show(10)
        
        # 2. 城市級銷售分析
        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城市銷售分析:")
        city_sales.show()
        
        # 3. 時間序列分析
        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每日銷售趨勢:")
        daily_sales.show(10)
        
        return {
            "user_behavior": user_behavior,
            "city_sales": city_sales, 
            "daily_sales": daily_sales
        }
    
    def window_function_analysis(self, transactions_df):
        """窗口函數分析"""
        print("\n=== 窗口函數分析 ===")
        
        from pyspark.sql.window import Window
        
        # 定義窗口規(guī)范
        user_window = Window \
            .partitionBy("user_id") \
            .orderBy("timestamp") \
            .rowsBetween(Window.unboundedPreceding, Window.currentRow)
        
        # 使用窗口函數計算累計值
        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("用戶累計消費分析:")
        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進行高級分析"""
        print("\n=== 使用Pandas UDF進行高級分析 ===")
        
        from pyspark.sql.functions import pandas_udf
        from pyspark.sql.types import DoubleType
        
        # 定義Pandas UDF計算用戶價值評分
        @pandas_udf(DoubleType())
        def calculate_customer_value(transaction_counts, total_spent, unique_categories):
            """計算客戶價值評分"""
            # 使用RFM-like評分機制
            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()
            
            # 綜合評分（加權平均）
            overall_score = 0.4 * frequency_score + 0.4 * monetary_score + 0.2 * variety_score
            return overall_score
        
        # 準備數據
        user_metrics = transactions_df \
            .groupBy("user_id") \
            .agg(
                count("transaction_id").alias("transaction_count"),
                sum("amount").alias("total_spent"),
                countDistinct("category").alias("unique_categories")
            )
        
        # 應用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("客戶價值分析:")
        user_value_analysis.select("user_id", "city", "transaction_count", 
                                 "total_spent", "customer_value_score").show(10)
        
        return user_value_analysis

# 高級分析演示
if spark:
    analyzer = AdvancedDataAnalyzer(spark)
    
    # 執(zhí)行復雜分析
    analysis_results = analyzer.perform_complex_aggregations(
        cleaned_data["users"], cleaned_data["transactions"]
    )
    
    # 窗口函數分析
    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):
        """分析查詢執(zhí)行計劃"""
        print(f"\n=== {description} 執(zhí)行計劃分析 ===")
        
        # 顯示邏輯計劃
        print("邏輯執(zhí)行計劃:")
        print(df._jdf.queryExecution().logical().toString())
        
        # 顯示物理計劃  
        print("\n物理執(zhí)行計劃:")
        print(df._jdf.queryExecution().executedPlan().toString())
        
        # 顯示優(yōu)化計劃
        print("\n優(yōu)化后的執(zhí)行計劃:")
        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í)行時間: {time1:.4f}秒")
        print(f"緩存后執(zhí)行時間: {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} ===")
        
        # 檢查當前分區(qū)數
        initial_partitions = df.rdd.getNumPartitions()
        print(f"初始分區(qū)數: {initial_partitions}")
        
        # 重新分區(qū)
        optimized_df = df.repartition(200, partition_col)
        optimized_partitions = optimized_df.rdd.getNumPartitions()
        print(f"優(yōu)化后分區(qū)數: {optimized_partitions}")
        
        # 顯示分區(qū)統(tǒng)計
        partition_stats = optimized_df \
            .groupBy(spark_partition_id().alias("partition_id")) \
            .count() \
            .orderBy("partition_id")
        
        print("分區(qū)數據分布:")
        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
        
        # 標準連接
        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"標準連接 - 記錄數: {standard_count:,}, 時間: {standard_time:.2f}秒")
        print(f"廣播連接 - 記錄數: {broadcast_count:,}, 時間: {broadcast_time:.2f}秒")
        print(f"性能提升: {standard_time/broadcast_time:.2f}x")
        
        return broadcast_join

# 性能優(yōu)化演示
if spark:
    optimizer = PerformanceOptimizer(spark)
    
    # 分析執(zhí)行計劃
    sample_df = cleaned_data["transactions"].filter(col("amount") > 50)
    optimizer.analyze_query_plan(sample_df, "過濾交易數據")
    
    # 緩存策略演示
    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
電商用戶行為分析系統(tǒng) - 完整PySpark實現
"""

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:
    """電商用戶行為分析系統(tǒng)"""
    
    def __init__(self):
        self.spark = self.initialize_spark_session()
        self.analysis_results = {}
    
    def initialize_spark_session(self):
        """初始化Spark會話"""
        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會話初始化完成")
        return spark
    
    def generate_ecommerce_data(self, num_users=100000, num_transactions=1000000):
        """生成電商模擬數據"""
        print("生成電商模擬數據...")
        
        # 用戶數據
        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)
        
        # 交易數據
        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"? 生成用戶數據: {users_df.count():,} 條")
        print(f"? 生成交易數據: {transactions_df.count():,} 條")
        
        return users_df, transactions_df
    
    def comprehensive_user_analysis(self, users_df, transactions_df):
        """綜合用戶行為分析"""
        print("\n" + "="*50)
        print("開始綜合用戶行為分析")
        print("="*50)
        
        # 1. 用戶基本行為分析
        user_behavior = self.analyze_user_behavior(users_df, transactions_df)
        
        # 2. RFM分析
        rfm_analysis = self.perform_rfm_analysis(transactions_df)
        
        # 3. 用戶聚類分析
        user_clusters = self.perform_user_clustering(user_behavior)
        
        # 4. 時間序列分析
        time_analysis = self.analyze_temporal_patterns(transactions_df)
        
        # 5. 生成業(yè)務洞察
        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):
        """用戶行為分析"""
        print("進行用戶行為分析...")
        
        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"? 用戶行為分析完成，分析 {user_behavior.count():,} 名用戶")
        return user_behavior
    
    def perform_rfm_analysis(self, transactions_df):
        """RFM分析（最近購買、購買頻率、購買金額）"""
        print("進行RFM分析...")
        
        # 計算基準日期（最近30天）
        max_date = transactions_df.select(max("timestamp")).first()[0]
        baseline_date = max_date - timedelta(days=30)
        
        # RFM計算
        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"),  # 最近購買
                count("transaction_id").alias("frequency"),  # 購買頻率
                sum("amount").alias("monetary")  # 購買金額
            ) \
            .filter(col("frequency") > 0)  # 只分析有購買行為的用戶
        
        # RFM評分（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, "冠軍客戶")
                       .when(col("rfm_score") >= 9, "忠實客戶") 
                       .when(col("rfm_score") >= 6, "潛力客戶")
                       .when(col("rfm_score") >= 3, "新客戶")
                       .otherwise("流失風險客戶"))
        
        print(f"? RFM分析完成，分析 {rfm_scored.count():,} 名用戶")
        return rfm_scored
    
    def perform_user_clustering(self, user_behavior):
        """用戶聚類分析"""
        print("進行用戶聚類分析...")
        
        # 準備特征
        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)
        
        # 特征標準化
        scaler = StandardScaler(inputCol="features", outputCol="scaled_features")
        scaler_model = scaler.fit(feature_vector)
        scaled_data = scaler_model.transform(feature_vector)
        
        # K-means聚類
        kmeans = KMeans(featuresCol="scaled_features", k=4, seed=42)
        kmeans_model = kmeans.fit(scaled_data)
        clustered_data = kmeans_model.transform(scaled_data)
        
        # 評估聚類效果
        evaluator = ClusteringEvaluator(featuresCol="scaled_features")
        silhouette_score = evaluator.evaluate(clustered_data)
        
        print(f"? 用戶聚類完成，輪廓系數: {silhouette_score:.3f}")
        
        # 分析聚類特征
        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("聚類特征分析:")
        cluster_profiles.show()
        
        return {
            "clustered_data": clustered_data,
            "cluster_profiles": cluster_profiles,
            "silhouette_score": silhouette_score
        }
    
    def analyze_temporal_patterns(self, transactions_df):
        """時間序列模式分析"""
        print("進行時間序列分析...")
        
        # 按小時分析購買模式
        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")
        
        # 按星期分析購買模式
        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")
        
        # 月度趨勢分析
        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("? 時間序列分析完成")
        
        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è)務洞察"""
        print("生成業(yè)務洞察...")
        
        insights = {}
        
        # 1. 關鍵指標
        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細分統(tǒng)計
        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. 聚類分析洞察
        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. 時間模式洞察
        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è)務洞察生成完成")
        return insights
    
    def generate_comprehensive_report(self):
        """生成綜合分析報告"""
        if not self.analysis_results:
            print("請先執(zhí)行分析")
            return
        
        insights = self.analysis_results["business_insights"]
        
        print("\n" + "="*60)
        print("電商用戶行為分析報告")
        print("="*60)
        
        # 關鍵指標
        print("\n?? 關鍵業(yè)務指標:")
        metrics = insights["key_metrics"]
        print(f"  ? 總用戶數: {metrics['total_users']:,}")
        print(f"  ? 總營收: ￥{metrics['total_revenue']:,.2f}")
        print(f"  ? 平均交易價值: ￥{metrics['avg_transaction_value']:.2f}")
        print(f"  ? 平均客戶評分: {metrics['avg_customer_rating']:.2f}/5")
        
        # RFM細分
        print("\n?? RFM客戶細分:")
        for segment, count in insights["rfm_segments"].items():
            percentage = (count / metrics['total_users']) * 100
            print(f"  ? {segment}: {count:,} 人 ({percentage:.1f}%)")
        
        # 聚類分析
        print("\n?? 用戶聚類分析:")
        for cluster in insights["cluster_analysis"]:
            print(f"  ? 聚類{cluster['cluster_id']}: {cluster['size']:,} 用戶")
            print(f"    平均交易數: {cluster['avg_transactions']:.1f}")
            print(f"    平均消費: ￥{cluster['avg_spent']:,.2f}")
        
        # 時間洞察
        print("\n? 時間模式洞察:")
        temporal = insights["temporal_insights"]
        print(f"  ? 高峰時段: {temporal['peak_hour']}:00 ({temporal['peak_hour_transactions']} 筆交易)")
        print(f"  ? 最繁忙日期: {temporal['busiest_day']}")
        
        # 性能指標
        if "silhouette_score" in self.analysis_results["user_clusters"]:
            score = self.analysis_results["user_clusters"]["silhouette_score"]
            print(f"\n?? 聚類質量: {score:.3f} (輪廓系數)")
    
    def save_analysis_results(self, output_path):
        """保存分析結果"""
        print(f"\n保存分析結果到: {output_path}")
        
        try:
            # 保存用戶行為數據
            self.analysis_results["user_behavior"] \
                .write \
                .mode("overwrite") \
                .parquet(f"{output_path}/user_behavior")
            
            # 保存RFM分析結果
            self.analysis_results["rfm_analysis"] \
                .write \
                .mode("overwrite") \
                .parquet(f"{output_path}/rfm_analysis")
            
            # 保存聚類結果
            self.analysis_results["user_clusters"]["clustered_data"] \
                .write \
                .mode("overwrite") \
                .parquet(f"{output_path}/user_clusters")
            
            print("? 分析結果保存完成")
            
        except Exception as e:
            print(f"? 保存失敗: {e}")
    
    def stop(self):
        """停止Spark會話"""
        self.spark.stop()
        print("? Spark會話已停止")

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

if __name__ == "__main__":
    main()

class ProductionDeployment:
    """生產環(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)建生產環(huán)境Spark會話"""
        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()

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

class OptimizationChecklist:
    """性能優(yōu)化檢查清單"""
    
    @staticmethod
    def get_checklist():
        """獲取優(yōu)化檢查清單"""
        return {
            "數據讀取": [
                "使用列式存儲格式(Parquet/ORC)",
                "合理設置分區(qū)數",
                "使用謂詞下推優(yōu)化"
            ],
            "數據處理": [
                "避免不必要的shuffle操作",
                "使用廣播連接小表",
                "合理使用緩存策略",
                "盡早過濾不需要的數據"
            ],
            "內存管理": [
                "監(jiān)控Executor內存使用",
                "合理設置序列化器",
                "避免數據傾斜",
                "使用堆外內存"
            ],
            "執(zhí)行優(yōu)化": [
                "啟用自適應查詢執(zhí)行",
                "合理設置并行度",
                "使用向量化UDF",
                "優(yōu)化數據本地性"
            ]
        }
    
    @staticmethod
    def validate_configuration(spark_conf):
        """驗證配置合理性"""
        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}")