3. Spark批处理基础：核心概念与作业提交流程

在风控特征工程中，每日需处理数亿至数十亿条用户行为日志、交易流水与外部数据。Spark作为分布式批处理引擎，是特征分层加工的核心计算平台。本章聚焦Spark RDD与DataFrame两大抽象、SparkSession的初始化规范以及批处理作业的完整提交流程，为后续特征计算Pipeline打下基础。

3.1 Spark核心抽象：RDD与DataFrame

3.2 SparkSession初始化规范

SparkSession是Spark 2.0+的统一入口，替代了旧的SparkContext、SQLContext、HiveContext。在风控生产环境中，初始化需考虑资源配置、Hive集成与动态分配。

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("risk_feature_batch") \
    .config("spark.sql.adaptive.enabled", "true") \
    .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \
    .config("spark.sql.adaptive.coalescePartitions.minPartitionNum", "200") \
    .config("spark.sql.shuffle.partitions", "400") \
    .config("spark.dynamicAllocation.enabled", "true") \
    .config("spark.dynamicAllocation.minExecutors", "10") \
    .config("spark.dynamicAllocation.maxExecutors", "200") \
    .config("spark.sql.warehouse.dir", "/user/hive/warehouse") \
    .enableHiveSupport() \
    .getOrCreate()

3.3 批处理作业提交流程

一个完整的Spark批处理作业从提交到完成，经历以下阶段：

1. 客户端提交：使用spark-submit脚本，指定主类、部署模式、资源配置与依赖JAR/Py文件。
2. Driver启动：在集群的Gateway节点（client模式）或YARN ApplicationMaster（cluster模式）上启动Driver进程。
3. SparkContext初始化：Driver创建SparkContext，向Cluster Manager（YARN/K8s）申请资源。
4. Executor分配：Cluster Manager在Worker节点上启动Executor进程，并向Driver注册。
5. 作业调度：Driver将用户代码转换为DAG（有向无环图），由DAGScheduler拆分为Stage，TaskScheduler将Task分发到Executor。
6. 任务执行：Executor并行执行Task，数据在Shuffle阶段跨节点传输。
7. 结果收集与清理：Driver收集最终结果（如保存到Hive表），释放所有Executor资源。

spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --queue risk_priority \
  --num-executors 100 \
  --executor-cores 4 \
  --executor-memory 8g \
  --driver-memory 4g \
  --conf spark.sql.adaptive.enabled=true \
  --py-files feature_lib.zip,config.py \
  risk_feature_pipeline.py \
  --date 2025-03-01 \
  --feature_type rolling_30d

3.4 核心流程SVG图

3.5 RDD与DataFrame血缘关系SVG图

3.6 风控批处理最佳实践

# 1. 读取原始日志（按分区过滤）
raw_df = spark.table("ods.user_behavior_log") \
    .where(col("dt") == batch_date)

# 2. 特征加工（窗口聚合）
feature_df = raw_df.groupBy("user_id") \
    .agg(
        count("*").alias("total_actions_30d"),
        sum(when(col("action_type") == "login", 1).otherwise(0)).alias("login_cnt_30d"),
        avg(col("amount")).alias("avg_amount_30d")
    )

# 3. 关联维度表（广播）
dim_df = spark.table("dim.user_risk_level").hint("broadcast")
result_df = feature_df.join(dim_df, "user_id", "left")

# 4. 写入特征宽表
result_df.write \
    .mode("overwrite") \
    .partitionBy("dt") \
    .format("parquet") \
    .saveAsTable("feature.user_rolling_features")