AI数据分析训练营实战：基于腾讯云EMR与PySpark的大规模用户行为预测系统

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, when, count, isnan, isnull, udf
from pyspark.sql.types import *
import os

spark = SparkSession.builder \
    .appName("UserPurchasePrediction") \
    .config("spark.sql.adaptive.enabled", "true") \
    .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \
    .config("spark.sql.parquet.compression.codec", "snappy") \
    .getOrCreate()

# 读取COS上的数据（使用cosn://协议）
base_path = "cosn://ecommerce-logs/raw/"
train_dates = [f"date=2026-08-{str(i).zfill(2)}" for i in range(8, 15)]  # 8-14日
val_dates = ["date=2026-08-15"]

def read_partitions(dates):
    paths = [os.path.join(base_path, d) for d in dates]
    return spark.read.parquet(*paths)

train_df = read_partitions(train_dates)
val_df = read_partitions(val_dates)

print(f"Train records: {train_df.count()}, Val records: {val_df.count()}")

from pyspark.sql.functions import lit, rand, expr, sum as _sum, count as _count, avg, stddev

# 1. 添加随机盐（0~N-1）
salt_factor = 10
window_start = "2026-08-14 00:00:00"  # 预测日之前

# 过滤预测日之前的行为
history_df = train_df.filter(col("event_time") < window_start)

# 加盐
salted_df = history_df.withColumn("salt", (rand() * salt_factor).cast("int"))

# 2. 先按 (user_id, salt) 做局部聚合
local_agg = salted_df.groupBy("user_id", "salt").agg(
    _count("event_id").alias("cnt_local"),
    _sum("pay_price").alias("sum_price_local"),
    avg("stay_duration").alias("avg_duration_local")
)

# 3. 再按 user_id 做全局合并
user_stats = local_agg.groupBy("user_id").agg(
    _sum("cnt_local").alias("user_total_actions"),
    _sum("sum_price_local").alias("user_total_spent"),
    avg("avg_duration_local").alias("user_avg_duration")
)

# 处理缺失（用全局中位数填充，这里用均值简化）
global_avg = user_stats.select(avg("user_total_actions")).collect()[0][0]
user_stats = user_stats.fillna({"user_total_actions": global_avg, 
                                "user_total_spent": 0.0, 
                                "user_avg_duration": 120.0})

from pyspark.sql import Window
from pyspark.sql.functions import mean, variance, when, lit

# 计算类别出现频率
category_count = train_df.groupBy("category_id").agg(count("*").alias("cat_freq"))
total_count = train_df.count()
category_encoding = category_count.withColumn("cat_freq_ratio", col("cat_freq") / total_count)

# 目标编码（使用全局转化率作为先验，平滑因子m=100）
global_ctr = train_df.filter(col("label") == 1).count() / total_count

# 计算每个类别的转化率
category_ctr = train_df.groupBy("category_id").agg(
    (sum("label") / count("*")).alias("raw_ctr"),
    count("*").alias("cnt")
)
# 平滑：CTR_encoded = (raw_ctr * cnt + global_ctr * m) / (cnt + m)
smooth_ctr = category_ctr.withColumn(
    "cat_target_enc",
    (col("raw_ctr") * col("cnt") + lit(global_ctr) * lit(100)) / (col("cnt") + lit(100))
)

# 将编码信息广播Join到全量数据（广播小表）
from pyspark.sql.functions import broadcast
train_df_encoded = train_df.join(broadcast(smooth_ctr.select("category_id", "cat_target_enc")), 
                                  on="category_id", how="left")
train_df_encoded = train_df_encoded.join(broadcast(category_encoding.select("category_id", "cat_freq_ratio")),
                                          on="category_id", how="left")

from pyspark.sql.functions import hour, dayofweek, unix_timestamp, from_unixtime

def cyclic_encoding(df, timestamp_col):
    df = df.withColumn("hour_of_day", hour(col(timestamp_col)))
    df = df.withColumn("day_of_week", dayofweek(col(timestamp_col)))
    # 正弦/余弦
    df = df.withColumn("hour_sin", sin((col("hour_of_day") * 2 * 3.14159) / 24))
    df = df.withColumn("hour_cos", cos((col("hour_of_day") * 2 * 3.14159) / 24))
    df = df.withColumn("dow_sin", sin((col("day_of_week") * 2 * 3.14159) / 7))
    df = df.withColumn("dow_cos", cos((col("day_of_week") * 2 * 3.14159) / 7))
    return df

train_df_feat = cyclic_encoding(train_df_encoded, "event_time")
val_df_feat = cyclic_encoding(val_df_encoded, "event_time")

numeric_cols = ["user_total_actions", "user_total_spent", "user_avg_duration", 
                "page_view_depth", "scroll_ratio", "device_battery"]
categorical_enc_cols = ["cat_target_enc", "cat_freq_ratio", "brand_target_enc"]
time_cols = ["hour_sin", "hour_cos", "dow_sin", "dow_cos"]

feature_cols = numeric_cols + categorical_enc_cols + time_cols + ["price", "discount_rate"]

# 将Spark DataFrame转换为向量化的特征（使用VectorAssembler）
from pyspark.ml.feature import VectorAssembler, StandardScaler
from pyspark.ml import Pipeline

assembler = VectorAssembler(inputCols=feature_cols, outputCol="raw_features")
scaler = StandardScaler(inputCol="raw_features", outputCol="scaled_features", 
                        withStd=True, withMean=True)

# 构建Pipeline仅用于转换（不包含模型，模型用XGBoost单独训练）
pipeline = Pipeline(stages=[assembler, scaler])
pipeline_model = pipeline.fit(train_df_feat)  # 在训练集上计算均值/方差

train_transformed = pipeline_model.transform(train_df_feat).select("scaled_features", "label")
val_transformed = pipeline_model.transform(val_df_feat).select("scaled_features", "label")

from xgboost.spark import SparkXGBClassifier
from pyspark.ml.evaluation import BinaryClassificationEvaluator

# 初始基线模型
xgb = SparkXGBClassifier(
    features_col="scaled_features",
    label_col="label",
    num_workers=4,          # 利用4个Executor
    num_round=100,
    max_depth=6,
    eta=0.1,
    subsample=0.8,
    colsample_bytree=0.8,
    objective="binary:logistic",
    eval_metric="auc",
    early_stopping_rounds=10,
    missing=0.0,
    use_gpu=False,          # EMR CPU集群
    nthread=16
)

# 训练
model = xgb.fit(train_transformed)

from hyperopt import fmin, tpe, hp, STATUS_OK, Trials
from hyperopt.spark import SparkTrials

# 定义搜索空间
space = {
    'max_depth': hp.choice('max_depth', [4, 6, 8, 10]),
    'eta': hp.uniform('eta', 0.01, 0.3),
    'subsample': hp.uniform('subsample', 0.6, 1.0),
    'colsample_bytree': hp.uniform('colsample_bytree', 0.6, 1.0),
    'min_child_weight': hp.choice('min_child_weight', [1, 3, 5]),
    'scale_pos_weight': hp.uniform('scale_pos_weight', 0.5, 3.0)  # 处理类别不平衡
}

def objective(params):
    clf = SparkXGBClassifier(
        features_col="scaled_features", label_col="label",
        num_workers=4, num_round=150,
        max_depth=params['max_depth'],
        eta=params['eta'],
        subsample=params['subsample'],
        colsample_bytree=params['colsample_bytree'],
        min_child_weight=params['min_child_weight'],
        scale_pos_weight=params['scale_pos_weight'],
        early_stopping_rounds=10,
        objective="binary:logistic", eval_metric="auc",
        missing=0.0, use_gpu=False, nthread=16
    )
    # 使用3折交叉验证（Spark XGBoost支持CV）
    cv_model = clf.fit(train_transformed)  # 此处可改为crossValidate，为了加速我们直接训练并验证集评估
    # 实际调参时应在验证集上评估，此处简化
    preds = cv_model.transform(val_transformed)
    evaluator = BinaryClassificationEvaluator(labelCol="label", metricName="areaUnderROC")
    auc = evaluator.evaluate(preds)
    return {'loss': -auc, 'status': STATUS_OK}

# 并行执行（4个并发）
spark_trials = SparkTrials(parallelism=4)
best_params = fmin(fn=objective, space=space, algo=tpe.suggest, 
                   max_evals=20, trials=spark_trials)

print("Best params:", best_params)
# 用最佳参数重新训练最终模型
best_model = SparkXGBClassifier(
    features_col="scaled_features", label_col="label",
    num_workers=4, num_round=200,
    max_depth=best_params['max_depth'],
    eta=best_params['eta'],
    subsample=best_params['subsample'],
    colsample_bytree=best_params['colsample_bytree'],
    min_child_weight=best_params['min_child_weight'],
    scale_pos_weight=best_params['scale_pos_weight'],
    early_stopping_rounds=15,
    objective="binary:logistic", eval_metric="auc",
    missing=0.0
).fit(train_transformed)

# 验证集AUC
val_preds = best_model.transform(val_transformed)
evaluator = BinaryClassificationEvaluator(labelCol="label", metricName="areaUnderROC")
val_auc = evaluator.evaluate(val_preds)
print(f"Validation AUC: {val_auc:.4f}")

# 特征重要性（XGBoost内置）
importance = best_model._java_model.getFeatureScore()
# 转为Python dict
import json
importance_dict = json.loads(importance.toString())
# 按分值排序
sorted_imp = sorted(importance_dict.items(), key=lambda x: x[1], reverse=True)
print("Top 10 important features:", sorted_imp[:10])

# 保存pipeline（包含assembler和scaler）
pipeline_model.write().overwrite().save("cosn://ecommerce-models/pipeline_model")

# XGBoost模型原生保存（使用内置save）
best_model.write().overwrite().save("cosn://ecommerce-models/xgb_model")

# 额外保存特征列名和类别映射（用于SCF侧）
feature_names = feature_cols  # 47个
import pickle
with open("feature_names.pkl", "wb") as f:
    pickle.dump(feature_names, f)
# 上传至COS（使用hadoop fs）
spark.sparkContext._jsc.hadoopConfiguration().set("fs.cosn.impl", "org.apache.hadoop.fs.CosNFileSystem")
# 通过SparkContext上传
spark.sparkContext.addFile("feature_names.pkl")

# index.py (SCF Python runtime)
import os
import pickle
import numpy as np
import xgboost as xgb
from qcloud_cos import CosConfig, CosS3Client
import logging
import json

logger = logging.getLogger()
logger.setLevel(logging.INFO)

# 环境变量
COS_REGION = os.environ.get('COS_REGION', 'ap-guangzhou')
COS_BUCKET = os.environ.get('COS_BUCKET', 'ecommerce-models-123456789')
MODEL_KEY = 'xgb_model/model.json'  # XGBoost JSON格式
PIPELINE_KEY = 'pipeline_model'     # 包含scaler
FEATURE_NAMES_KEY = 'feature_names.pkl'

# 全局缓存
_model = None
_scaler = None
_feature_names = None

def load_models():
    global _model, _scaler, _feature_names
    if _model is not None:
        return
    # 初始化COS客户端（内网访问）
    config = CosConfig(Region=COS_REGION, SecretId=os.environ['TENCENTCLOUD_SECRETID'],
                       SecretKey=os.environ['TENCENTCLOUD_SECRETKEY'])
    client = CosS3Client(config)
    
    # 下载特征名称
    resp = client.get_object(Bucket=COS_BUCKET, Key=FEATURE_NAMES_KEY)
    _feature_names = pickle.loads(resp['Body'].get_raw_stream().read())
    
    # 下载XGBoost模型（JSON格式）
    resp = client.get_object(Bucket=COS_BUCKET, Key=MODEL_KEY)
    model_bytes = resp['Body'].get_raw_stream().read()
    import tempfile
    with tempfile.NamedTemporaryFile(delete=False, suffix='.json') as f:
        f.write(model_bytes)
        model_path = f.name
    _model = xgb.XGBClassifier()
    _model.load_model(model_path)
    os.remove(model_path)
    
    # 下载Pipeline（包含scaler）——由于pipeline包含Java对象，SCF侧无法直接使用Spark的Scaler，我们手动用numpy实现标准化
    # 从COS读取均值/方差（提前在训练时存为JSON）
    resp = client.get_object(Bucket=COS_BUCKET, Key='scaler_stats.json')
    stats = json.loads(resp['Body'].get_raw_stream().read())
    _scaler = {
        'mean': np.array(stats['mean']),
        'std': np.array(stats['std'])
    }
    logger.info("Models loaded successfully")

def preprocess(raw_features_dict):
    # 根据_feature_names构建有序向量
    vec = np.zeros(len(_feature_names))
    for i, name in enumerate(_feature_names):
        vec[i] = raw_features_dict.get(name, 0.0)
    # 标准化
    vec = (vec - _scaler['mean']) / (_scaler['std'] + 1e-8)
    return vec.reshape(1, -1)

def main_handler(event, context):
    try:
        load_models()
        # 解析请求（假设POST JSON）
        body = json.loads(event['body'])
        user_features = body['features']  # dict
        vec = preprocess(user_features)
        prob = _model.predict_proba(vec)[0][1]  # 正类概率
        return {
            'statusCode': 200,
            'body': json.dumps({'prediction': float(prob), 'success': True})
        }
    except Exception as e:
        logger.error(str(e))
        return {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }

# 在训练脚本中
scaler_model = pipeline_model.stages[1]  # StandardScaler
mean_arr = scaler_model.mean.toArray().tolist()
std_arr = scaler_model.std.toArray().tolist()
import json
with open("scaler_stats.json", "w") as f:
    json.dump({"mean": mean_arr, "std": std_arr}, f)
# 上传至COS
spark.sparkContext.addFile("scaler_stats.json")
# 使用hadoop fs -put 或 sdk上传

curl -X POST https://api-gateway-url/predict \
  -H "Content-Type: application/json" \
  -d '{"features": {"user_total_actions": 12, "price": 99.9, ...}}'
# 返回 {"prediction":0.82, "success":true}

# 对广播小表做repartition
user_stats_broadcast = spark.sparkContext.broadcast(user_stats.collectAsMap())
# 使用UDF进行map-side join，避免shuffle
def get_user_stat(user_id):
    return user_stats_broadcast.value.get(user_id, (0, 0.0, 120.0))
udf_get = udf(get_user_stat, StructType([...]))