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)勢": [
                "內(nèi)存計算比Hadoop MapReduce快100倍",
                "基于DAG的優(yōu)化執(zhí)行引擎",
                "懶加載機(jī)制優(yōu)化計算流程"
            ],
            "易用性優(yōu)勢": [
                "Python簡潔的API接口",
                "與Pandas無縫集成",
                "豐富的機(jī)器學(xué)習(xí)庫"
            ],
            "生態(tài)系統(tǒng)": [
                "Spark SQL: 結(jié)構(gòu)化數(shù)據(jù)處理",
                "Spark Streaming: 實時數(shù)據(jù)處理", 
                "MLlib: 機(jī)器學(xué)習(xí)庫",
                "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 = {
            "數(shù)據(jù)大小(GB)": data_sizes,
            "Pandas處理時間(秒)": pandas_times,
            "PySpark處理時間(秒)": 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ū)動節(jié)點(diǎn)(Driver)": "執(zhí)行main方法，創(chuàng)建SparkContext",
            "集群管理器(Cluster Manager)": "資源分配和調(diào)度",
            "工作節(jié)點(diǎn)(Worker Node)": "執(zhí)行具體計算任務(wù)", 
            "執(zhí)行器(Executor)": "在工作節(jié)點(diǎn)上運(yùn)行任務(wù)"
        }
        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"應(yī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會話創(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會話"""
        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轉(zhuǎn)換與行動操作演示 ===")
        
        # 創(chuàng)建RDD
        data = list(range(1, 101))
        rdd = self.sc.parallelize(data, 4)  # 4個分區(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))
        
        # 行動操作 - 觸發(fā)計算
        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)換操作不會立即執(zhí)行
        mapped_rdd = rdd.map(lambda x: x * 2)
        filtered_rdd = mapped_rdd.filter(lambda x: x > 10)
        
        print("轉(zhuǎn)換操作定義完成，尚未執(zhí)行計算")
        print("觸發(fā)行動操作...")
        
        # 行動操作觸發(fā)計算
        result = filtered_rdd.collect()
        print(f"計算結(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: 通過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ī)模數(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文件（列式存儲，更適合大數(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ù)集...")
        
        # 用戶數(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"生成用戶數(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 = {}
        
        # 用戶數(shù)據(jù)清洗
        users_df = df_dict["users"]
        print(f"原始用戶數(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"清洗后用戶數(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ì)量報告
        self.generate_data_quality_report(cleaned_data)
        
        return cleaned_data
    
    def generate_data_quality_report(self, data_dict):
        """生成數(shù)據(jù)質(zhì)量報告"""
        print("\n=== 數(shù)據(jù)質(zhì)量報告 ===")
        
        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} 數(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ù)（在實際應(yīng)用中替換為真實數(shù)據(jù)路徑）
    raw_data = processor.generate_sample_data(50000)
    
    # 數(shù)據(jù)清洗
    cleaned_data = processor.comprehensive_data_cleaning(raw_data)

class AdvancedDataAnalyzer:
    """高級數(shù)據(jù)分析器"""
    
    def __init__(self, spark):
        self.spark = spark
    
    def perform_complex_aggregations(self, users_df, transactions_df):
        """執(zhí)行復(fù)雜聚合分析"""
        print("=== 復(fù)雜聚合分析 ===")
        
        # 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):
        """窗口函數(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ù)計算累計值
        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("用戶累計消費(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)行高級分析"""
        print("\n=== 使用Pandas UDF進(jìn)行高級分析 ===")
        
        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評分機(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()
            
            # 綜合評分（加權(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("客戶價值分析:")
        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í)行復(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):
        """分析查詢執(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} ===")
        
        # 檢查當(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)計
        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:,}, 時間: {standard_time:.2f}秒")
        print(f"廣播連接 - 記錄數(shù): {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, "過濾交易數(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
電商用戶行為分析系統(tǒng) - 完整PySpark實現(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:
    """電商用戶行為分析系統(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):
        """生成電商模擬數(shù)據(jù)"""
        print("生成電商模擬數(shù)據(jù)...")
        
        # 用戶數(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"? 生成用戶數(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):
        """綜合用戶行為分析"""
        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è)務(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):
        """用戶行為分析"""
        print("進(jìn)行用戶行為分析...")
        
        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("進(jìn)行RFM分析...")
        
        # 計算基準(zhǔn)日期（最近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("流失風(fēng)險客戶"))
        
        print(f"? RFM分析完成，分析 {rfm_scored.count():,} 名用戶")
        return rfm_scored
    
    def perform_user_clustering(self, user_behavior):
        """用戶聚類分析"""
        print("進(jìn)行用戶聚類分析...")
        
        # 準(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聚類
        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"? 用戶聚類完成，輪廓系數(shù): {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("進(jìn)行時間序列分析...")
        
        # 按小時分析購買模式
        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è)務(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)計
        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è)務(wù)洞察生成完成")
        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)
        
        # 關(guān)鍵指標(biāo)
        print("\n?? 關(guān)鍵業(yè)務(wù)指標(biāo):")
        metrics = insights["key_metrics"]
        print(f"  ? 總用戶數(shù): {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細(xì)分
        print("\n?? RFM客戶細(xì)分:")
        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"    平均交易數(shù): {cluster['avg_transactions']:.1f}")
            print(f"    平均消費(fèi): ￥{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']}")
        
        # 性能指標(biāo)
        if "silhouette_score" in self.analysis_results["user_clusters"]:
            score = self.analysis_results["user_clusters"]["silhouette_score"]
            print(f"\n?? 聚類質(zhì)量: {score:.3f} (輪廓系數(shù))")
    
    def save_analysis_results(self, output_path):
        """保存分析結(jié)果"""
        print(f"\n保存分析結(jié)果到: {output_path}")
        
        try:
            # 保存用戶行為數(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")
            
            # 保存聚類結(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會話"""
        self.spark.stop()
        print("? Spark會話已停止")

def main():
    """主函數(shù)"""
    print("啟動電商用戶行為分析系統(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. 生成報告
        analyzer.generate_comprehensive_report()
        
        # 4. 保存結(jié)果（在實際環(huán)境中取消注釋）
        # analyzer.save_analysis_results("hdfs://path/to/output")
        
        print("\n?? 分析完成!")
        
    except Exception as e:
        print(f"? 分析過程中出現(xiàn)錯誤: {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會話"""
        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ù)讀取": [
                "使用列式存儲格式(Parquet/ORC)",
                "合理設(shè)置分區(qū)數(shù)",
                "使用謂詞下推優(yōu)化"
            ],
            "數(shù)據(jù)處理": [
                "避免不必要的shuffle操作",
                "使用廣播連接小表",
                "合理使用緩存策略",
                "盡早過濾不需要的數(shù)據(jù)"
            ],
            "內(nèi)存管理": [
                "監(jiān)控Executor內(nèi)存使用",
                "合理設(shè)置序列化器",
                "避免數(shù)據(jù)傾斜",
                "使用堆外內(nèi)存"
            ],
            "執(zhí)行優(yōu)化": [
                "啟用自適應(yīng)查詢執(zhí)行",
                "合理設(shè)置并行度",
                "使用向量化UDF",
                "優(yōu)化數(shù)據(jù)本地性"
            ]
        }
    
    @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}")