📚 课程目录第2章 Spark环境搭建与数据源接入

2. Spark环境搭建与数据源接入：集群模式选择、读取Hive/MySQL/Kafka数据源配置

在风控特征工程中，Spark环境的选择直接决定了特征计算的吞吐量与延迟。本章将聚焦于Spark集群模式的选型对比，以及三种核心数据源（Hive、MySQL、Kafka）的接入配置与调优要点。

2.1 Spark集群模式选择

风控场景通常面临高吞吐、低延迟、资源隔离三大挑战。下表对比了四种常见部署模式在风控场景下的适用性：

模式	资源管理	适用场景	风控推荐度
Local	单机多线程	开发调试、小样本验证	⭐（仅开发）
Standalone	Spark内置Master/Worker	中小规模离线批处理	⭐⭐
YARN	Hadoop YARN	企业级离线/近线特征生产	⭐⭐⭐⭐⭐
Kubernetes	K8s容器编排	实时特征、弹性伸缩、云原生	⭐⭐⭐⭐

生产建议： 大多数风控团队选择 YARN 模式，因为它与Hive/HDFS深度集成，且支持动态资源分配与队列隔离，适合多租户特征生产环境。

以下为YARN模式下提交Spark作业的典型配置示例：

spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --executor-memory 8g \
  --num-executors 20 \
  --executor-cores 4 \
  --conf spark.dynamicAllocation.enabled=true \
  --conf spark.shuffle.service.enabled=true \
  --class com.risk.feature.FeaturePipeline \
  feature-pipeline.jar

2.2 读取Hive数据源配置

Hive是风控特征分层中存储宽表与中间层特征的核心仓库。Spark读取Hive表时，需注意元数据同步与分区剪枝优化。

2.2.1 基础配置

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("HiveFeatureReader") \
    .config("hive.metastore.uris", "thrift://metastore-host:9083") \
    .config("spark.sql.warehouse.dir", "/user/hive/warehouse") \
    .config("spark.sql.hive.convertMetastoreParquet", "true") \
    .enableHiveSupport() \
    .getOrCreate()

# 读取Hive分区表
df = spark.sql("""
    SELECT 
        user_id,
        feature_1,
        feature_2,
        dt
    FROM risk_dw.feature_wide_table
    WHERE dt = '2025-03-01'
""")

2.2.2 分区剪枝与谓词下推

风控特征表通常按天分区，必须确保Spark能利用分区过滤。以下配置可显著提升扫描效率：

参数	推荐值	说明
`spark.sql.hive.metastorePartitionPruning`	`true`	启用元数据分区剪枝
`spark.sql.adaptive.coalescePartitions.enabled`	`true`	动态合并小分区
`spark.sql.hive.convertMetastoreParquet`	`true`	将Hive SerDe表转为Spark原生Parquet读取

经验： 在读取Hive宽表时，尽量只select需要的特征列，避免select * 导致大量列式存储的IO浪费。

2.3 读取MySQL数据源配置

MySQL常用于存储风控规则配置、黑名单、设备指纹等维度数据。Spark通过JDBC连接MySQL，需注意连接池与并发读取策略。

2.3.1 JDBC连接配置

df_mysql = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:mysql://mysql-host:3306/risk_db?useSSL=false&serverTimezone=Asia/Shanghai") \
    .option("dbtable", "blacklist_device") \
    .option("user", "risk_user") \
    .option("password", "encrypted_password") \
    .option("driver", "com.mysql.cj.jdbc.Driver") \
    .option("numPartitions", "8") \
    .option("partitionColumn", "id") \
    .option("lowerBound", "1") \
    .option("upperBound", "10000000") \
    .load()

2.3.2 并发读取与连接池调优

参数	推荐值	作用
`numPartitions`	8~16	控制并发读取的JDBC连接数
`partitionColumn`	自增主键	用于数据切分的列
`fetchSize`	`10000`	每次拉取的行数，减少网络往返
`connectionTimeout`	`30000`	连接超时时间（毫秒）

注意： 不要将MySQL作为大规模特征存储使用。建议只读取维度小表（百万级以内），大表特征应提前同步至Hive或HBase。

2.4 读取Kafka数据源配置

实时风控场景下，Kafka是流式特征接入的标配。Spark Structured Streaming 提供了exactly-once语义与背压机制。

2.4.1 结构化流读取Kafka

df_kafka = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "kafka-broker1:9092,kafka-broker2:9092") \
    .option("subscribe", "risk_feature_input") \
    .option("startingOffsets", "latest") \
    .option("maxOffsetsPerTrigger", "50000") \
    .option("failOnDataLoss", "false") \
    .load()

# 解析Kafka消息（假设value为JSON格式）
from pyspark.sql.functions import from_json, col, schema_of_json

json_schema = schema_of_json("""{"user_id":"","event_type":"","amount":0.0,"ts":""}""")
df_parsed = df_kafka \
    .select(from_json(col("value").cast("string"), json_schema).alias("data")) \
    .select("data.*")

2.4.2 关键参数说明

参数	推荐值	说明
`maxOffsetsPerTrigger`	50000~200000	控制每个trigger处理的最大消息数，防止背压
`startingOffsets`	`latest` / `earliest`	实时场景用latest，回溯用earliest
`failOnDataLoss`	`false`	避免Kafka topic删除导致作业失败
`kafka.group.id`	自定义	用于offset管理，建议按作业命名

2.5 数据源接入架构总览

下图展示了风控特征分层中三种数据源的接入位置与数据流向：

2.6 多数据源联合读取示例

在实际风控特征生产中，经常需要将Hive、MySQL、Kafka三路数据做关联。以下代码展示了如何在一个Pipeline中完成联合读取：

# 1. 读取Hive离线特征
df_hive = spark.sql("SELECT user_id, feature_a, feature_b FROM risk_dw.feature_wide WHERE dt='2025-03-01'")

# 2. 读取MySQL黑名单维度
df_blacklist = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:mysql://mysql-host:3306/risk_db") \
    .option("dbtable", "blacklist_device") \
    .option("user", "risk_user") \
    .option("password", "encrypted_pwd") \
    .option("numPartitions", "4") \
    .option("partitionColumn", "id") \
    .option("lowerBound", "1") \
    .option("upperBound", "5000000") \
    .load()

# 3. 读取Kafka实时事件流
df_kafka_stream = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "kafka-broker:9092") \
    .option("subscribe", "risk_event_input") \
    .option("maxOffsetsPerTrigger", "100000") \
    .load() \
    .selectExpr("CAST(value AS STRING) as json_str") \
    .select(from_json("json_str", event_schema).alias("event")) \
    .select("event.*")

# 4. 关联：离线特征 + 维度表（批处理）
df_joined_batch = df_hive.join(df_blacklist, on="user_id", how="left")

# 5. 关联：实时流 + 维度表（流-维关联）
df_joined_stream = df_kafka_stream \
    .join(df_blacklist, on="device_id", how="left")

最佳实践： 对于流-维关联，建议将MySQL维度表缓存为广播变量（spark.sql.autoBroadcastJoinThreshold 调大），避免频繁的JDBC查询。

2.7 本章小结

集群模式：生产环境首选YARN，兼顾资源隔离与动态分配；K8s适合云原生实时场景。
Hive接入：利用分区剪枝与谓词下推，避免全表扫描；只select必要列。
MySQL接入：通过JDBC并发读取，控制numPartitions与fetchSize；仅用于维度小表。
Kafka接入：使用Structured Streaming，配置maxOffsetsPerTrigger实现背压控制。
联合读取：批流一体架构中，合理选择关联策略（广播join vs 普通join）。