📚 课程目录第3章 Spark生态与AML场景

3. Spark生态与AML场景：Spark Core、Spark SQL、MLlib在反洗钱中的角色

在反洗钱（AML）对照实验场景中，Spark生态的三大核心组件——Spark Core、Spark SQL 和 MLlib——各自承担着不可替代的职责。理解它们的分工与协作，是构建高效、可复现的AML实验平台的基础。

3.1 三大组件的AML角色矩阵

组件	核心能力	AML场景职责	对照实验中的价值
Spark Core	分布式计算引擎、RDD抽象、任务调度	海量交易数据的ETL管道、数据清洗、异常日志处理	提供实验数据分片与并行处理的底层保障，确保对照组的计算资源隔离
Spark SQL	结构化数据处理、DataFrame/Dataset API、Catalyst优化器	交易明细查询、客户画像聚合、特征工程中的SQL化操作	支持实验组与对照组的SQL级联查询，便于快速验证特征逻辑
MLlib	分布式机器学习算法库、特征工程、模型评估	异常交易检测模型（孤立森林、逻辑回归）、客户风险评分	提供标准化的模型训练与评估接口，保证实验对比的公平性

    实战要点：在AML对照实验中，Spark Core负责“数据管道”，Spark SQL负责“特征工厂”，MLlib负责“模型车间”。三者通过DataFrame无缝衔接，避免数据在不同组件间反复序列化。

3.2 Spark Core：AML数据管道的基石

Spark Core的RDD（弹性分布式数据集）是AML数据处理的底层抽象。在反洗钱场景中，交易数据通常以日志或消息队列的形式流入，Spark Core负责以下关键任务：

数据清洗：过滤无效交易（如金额为负、时间戳异常），使用map和filter算子完成。
数据分区：按客户ID或交易日期进行Hash分区，确保同一客户的交易落在同一分区，避免后续聚合操作的数据倾斜。
容错与重试：通过RDD的血统（Lineage）机制，在节点故障时自动重算丢失的分区，保障AML实验的稳定性。

// 示例：使用Spark Core清洗交易日志
val rawRDD = sc.textFile("hdfs:///data/aml/transactions/2025-03-01/")
val cleanRDD = rawRDD
    .filter(line => line.split(",").length == 12)  // 过滤格式异常行
    .map(line => {
        val fields = line.split(",")
        Transaction(fields(0), fields(1).toDouble, fields(2).toLong)
    })
    .filter(tx => tx.amount > 0 && tx.amount < 1000000)  // 金额合理性校验

3.3 Spark SQL：特征工程的结构化引擎

Spark SQL将AML特征工程从繁琐的RDD操作中解放出来。分析师可以直接使用SQL或DataFrame API完成复杂的聚合与关联查询，大幅提升实验迭代效率。

AML特征类型	Spark SQL实现方式	对照实验中的应用
交易频率特征	`GROUP BY customer_id, window(tx_time, '1 hour')`	对比不同窗口大小对异常检测的影响
金额统计特征	`AVG(amount), STDDEV(amount), MAX(amount)`	验证均值与中位数在风险评分中的差异
关联账户特征	`JOIN` 交易表与账户关系表	测试是否加入关联账户特征能提升召回率

效率提示：在AML对照实验中，建议将特征工程逻辑封装为Spark SQL的临时视图（TempView），实验组和对照组通过切换不同的视图定义来快速对比特征效果，无需重写整个ETL流程。

3.4 MLlib：分布式模型训练与评估

MLlib提供了面向大规模数据的机器学习算法实现，在AML场景中，最常用的包括：

逻辑回归（Logistic Regression）：用于客户风险二分类，输出风险概率。
随机森林（Random Forest）：处理高维特征，适合AML中的非线性模式识别。
孤立森林（Isolation Forest）：专为异常检测设计，在AML对照实验中常用于对比基线模型。

// 示例：使用MLlib训练AML风险模型
import org.apache.spark.ml.classification.LogisticRegression

val lr = new LogisticRegression()
    .setMaxIter(100)
    .setRegParam(0.01)
    .setLabelCol("is_risky")
    .setFeaturesCol("features")

val model = lr.fit(trainingData)

// 在对照实验中，可以对比不同正则化参数的效果
val lrExp = new LogisticRegression().setRegParam(0.1)
val lrCtrl = new LogisticRegression().setRegParam(0.01)

3.5 Spark与Hadoop/Flume的集成

在真实的AML生产环境中，数据源往往是异构的。Spark需要与Hadoop生态中的HDFS、Flume等组件深度集成，构建端到端的数据管道。

3.6 集成配置与最佳实践

在AML对照实验中，Spark与Hadoop/Flume的集成需要关注以下关键配置：

集成组件	配置项	AML场景推荐值	说明
Spark + HDFS	`spark.hadoop.dfs.replication`	3	保证AML数据的高可用性，防止节点故障导致实验数据丢失
Spark + Flume	`spark.streaming.receiver.writeAheadLog.enable`	true	启用预写日志，确保实时交易数据在流处理中的Exactly-Once语义
Spark + HDFS	`spark.sql.adaptive.coalescePartitions.enabled`	true	动态合并小分区，优化AML特征工程中的Shuffle性能

避坑指南：在AML对照实验中，务必确保实验组和对照组使用相同的HDFS数据快照。如果数据在实验过程中被增量写入，会导致两组数据分布不一致，使对比结果失效。建议在实验开始前对HDFS数据做快照（Snapshot）或复制一份静态副本。

3.7 本章小结

Spark Core、Spark SQL和MLlib在AML对照实验中形成了清晰的职责分工：Core负责数据管道的稳定与容错，SQL负责特征工程的高效表达，MLlib负责模型训练的标准化。通过与Hadoop/Flume的集成，Spark能够构建从数据采集、存储到计算分析的完整AML处理链路，为对照实验提供可靠、可复现的数据与计算基础。