📚 课程目录第4章数据采集与ETL

4. 数据采集与ETL：实时流式采集与离线批量采集

在支付风控体系中，数据采集是特征工程的“第一公里”。风控样本的时效性、完整性和一致性直接决定了后续模型的效果。本章围绕实时流式采集（Kafka/Flink）与离线批量采集（Sqoop/DataX）两条主线，深入讲解数据清洗与标准化的工程实践。

4.1 实时流式采集：Kafka + Flink 架构

支付交易具有高并发、低延迟的特点，风控决策需要在毫秒级完成。实时流式采集通过消息队列与流计算引擎，将交易事件从产生到进入特征计算的时间窗口压缩至秒级以内。

4.1.1 核心组件与职责

组件	角色	关键配置
Kafka	分布式消息队列，解耦数据生产与消费	topic 分区数、副本因子、消息保留策略
Flink	流计算引擎，实现实时 ETL 与特征加工	checkpoint 间隔、状态后端、水位线策略
Schema Registry	管理 Avro/Protobuf 序列化 schema	兼容性策略（BACKWARD/FORWARD）

  工程要点：Kafka 分区数应大于等于 Flink 并行度，避免数据倾斜；建议开启幂等生产者与 acks=all 保证至少一次语义。

4.1.2 Flink 实时 ETL 核心代码片段

// 从 Kafka 消费交易事件，进行实时清洗与标准化
DataStream<TradeEvent> rawStream = env.addSource(
    new FlinkKafkaConsumer<>("pay_trade_raw", 
        new AvroDeserializationSchema(TradeEvent.class), 
        kafkaProps));

DataStream<StandardizedTrade> cleanStream = rawStream
    .filter(event -> event.getAmount() != null && event.getAmount() > 0)
    .map(new RichMapFunction<TradeEvent, StandardizedTrade>() {
        @Override
        public StandardizedTrade map(TradeEvent event) {
            // 统一金额单位为分，时区转为 UTC
            long amountInCents = event.getAmount().longValue();
            String utcTime = LocalDateTime.parse(event.getTradeTime())
                .atZone(ZoneId.of("Asia/Shanghai"))
                .withZoneSameInstant(ZoneOffset.UTC)
                .toString();
            return new StandardizedTrade(event.getTradeId(), 
                event.getUserId(), amountInCents, utcTime);
        }
    })
    .keyBy(trade -> trade.getUserId())
    .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
    .process(new TradeFeatureAggregator());

避坑指南：Flink 状态后端建议使用 RocksDB 以支持大状态；水位线（watermark）设置应比实际延迟大 2-3 秒，避免乱序数据导致窗口计算偏差。

4.2 离线批量采集：Sqoop 与 DataX

离线采集用于历史交易数据、用户画像、设备指纹等大规模批处理场景。Sqoop 和 DataX 是业界主流的异构数据源同步工具，适用于 MySQL、Hive、HDFS 之间的全量/增量同步。

4.2.1 工具对比与选型

维度	Sqoop	DataX
数据源支持	RDBMS ↔ HDFS/Hive	RDBMS、NoSQL、文件、消息队列等 30+ 数据源
增量同步	支持 --incremental append/lastmodified	通过 where 条件或时间戳字段实现
性能	基于 MapReduce，适合大表全量	多线程通道，单机吞吐高，适合中大规模
容错机制	Task 失败自动重试 3 次	通道级重试 + 脏数据记录
运维复杂度	依赖 Hadoop 环境	独立部署，轻量级

  选型建议：如果已有 Hadoop 生态且需要全量同步历史交易表（亿级），优先 Sqoop；如果需要多数据源异构同步且对运维友好，选择 DataX。

4.2.2 DataX 离线同步配置示例

{
  "job": {
    "content": [
      {
        "reader": {
          "name": "mysqlreader",
          "parameter": {
            "connection": [{
              "jdbcUrl": ["jdbc:mysql://10.0.1.10:3306/pay_db"],
              "table": ["trade_record"],
              "where": "trade_date >= '2025-01-01' AND trade_date < '2025-02-01'"
            }],
            "column": ["trade_id", "user_id", "amount", "trade_time", "risk_level"],
            "splitPk": "trade_id"
          }
        },
        "writer": {
          "name": "hdfswriter",
          "parameter": {
            "defaultFS": "hdfs://nameservice1",
            "path": "/user/pay_risk/trade_record/dt=202501",
            "fileType": "parquet",
            "writeMode": "nonConflict"
          }
        }
      }
    ],
    "setting": {
      "speed": {"channel": 8},
      "errorLimit": {"record": 100}
    }
  }
}

4.3 数据清洗与标准化

原始支付数据通常包含空值、异常值、格式不一致等问题。清洗与标准化是保证样本质量的核心环节，直接影响特征分布的稳定性。

4.3.1 清洗规则清单

清洗类型	常见问题	处理策略
缺失值	用户 ID 为空、金额字段为 null	丢弃（drop）或填充默认值（-1 / "unknown"）
异常值	交易金额为负数、时间戳为 1970-01-01	阈值过滤 + 业务规则校验
格式不一致	手机号含国家码/不含、金额单位分/元混用	正则提取 + 统一单位转换
重复数据	同一笔交易被重复上报	基于 trade_id + user_id 去重
敏感信息	明文卡号、身份证号	脱敏（SHA256 哈希或掩码）

4.3.2 标准化字段映射

不同业务线（支付、退款、转账）的字段命名和类型可能不同，需统一映射到风控标准 schema。

-- 标准化字段映射示例（Hive SQL）
CREATE TABLE dwd_pay_trade_standard (
    trade_id      STRING  COMMENT '统一交易ID',
    user_id       STRING  COMMENT '用户ID',
    amount_cent   BIGINT  COMMENT '金额，单位：分',
    trade_time_utc STRING COMMENT '交易时间，UTC格式',
    channel       STRING  COMMENT '渠道：APP/WEB/API',
    risk_tag      STRING  COMMENT '风控标签'
)
PARTITIONED BY (dt STRING)
STORED AS PARQUET;

INSERT OVERWRITE TABLE dwd_pay_trade_standard PARTITION (dt='2025-01-15')
SELECT 
    trade_id,
    user_id,
    CASE WHEN amount_unit = 'yuan' THEN amount * 100 ELSE amount END AS amount_cent,
    from_utc_timestamp(trade_time, 'UTC') AS trade_time_utc,
    upper(channel) AS channel,
    COALESCE(risk_tag, 'normal') AS risk_tag
FROM ods_pay_trade_raw
WHERE dt = '2025-01-15';

注意：金额单位统一为“分”是支付风控的行业惯例，避免浮点数精度问题；时间字段统一为 UTC 字符串，便于跨时区特征计算。

4.4 实时与离线数据融合策略

风控样本构建通常需要融合实时流数据（当前交易上下文）与离线批数据（历史画像、统计特征）。常见融合模式如下：

预关联模式：离线预先计算用户历史统计特征（如近 7 天交易次数），存入 KV 存储；实时流中通过 user_id 直接读取。
窗口关联模式：Flink 中通过 interval join 将实时交易流与近期的离线特征流（如设备指纹变更记录）关联。
异步 IO 模式：实时流中通过 AsyncIO 调用外部存储（HBase/Redis）获取离线特征，避免阻塞主线程。

4.5 本章小结

实时流式采集以 Kafka 作为缓冲层，Flink 完成清洗、标准化与特征聚合，满足风控毫秒级响应需求。
离线批量采集通过 Sqoop/DataX 实现历史数据同步，重点控制增量策略与数据倾斜。
数据清洗需覆盖缺失值、异常值、格式不一致、重复数据和敏感信息五大类；标准化统一字段类型、单位和时区。
实时与离线融合是样本构建的关键，推荐预关联 + 异步 IO 的组合模式以平衡时效性与特征丰富度。