4. 数据摄入与采集：实时流式接入（Kafka）、批量导入（Sqoop/Spark）、数据校验机制

数据摄入，说白了就是解决「数据怎么进湖」的问题。风控场景下，这个环节尤其敏感——你想想看，一笔交易从发生到进入湖里，中间任何一秒的延迟都可能意味着风险敞口。我见过不少团队，花了大半年搭湖，结果数据进不来，或者进来的是脏数据，那整个湖就废了。

这一章，我们重点聊三种主流接入方式：Kafka 实时流、Sqoop 批量拉、Spark 灵活导。最后再补一个数据校验机制——这部分是我个人觉得最容易被忽视，但恰恰是最关键的环节。

4.1 实时流式接入：Kafka 管道

风控场景里，实时性要求最高的就是交易事件、登录行为、设备指纹这些。Kafka 几乎是标配，原因很简单：高吞吐、低延迟、持久化。

我习惯把 Kafka 到 Lakehouse 的链路分成三层：

  • 生产者层：业务系统把事件发到 Kafka topic，比如 trade_event、login_log。
  • 消费者层：Spark Structured Streaming 或 Flink 从 Kafka 拉数据，做轻量清洗后写入 Delta Lake。
  • 存储层：Delta Lake 的 Change Data Feed 机制保证 Exactly-Once 语义。

这里有个关键点：offset 管理。我在项目中遇到过，消费者挂了重启后重复消费，导致风控指标翻倍。后来我们强制用 Kafka 自带的 __consumer_offsets 配合 Checkpoint，才彻底解决。

核心配置示例（Spark Structured Streaming + Kafka + Delta）：

val df = spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "broker1:9092,broker2:9092")
  .option("subscribe", "trade_event")
  .option("startingOffsets", "latest")
  .option("failOnDataLoss", "false")
  .load()

df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
  .writeStream
  .format("delta")
  .option("checkpointLocation", "/lakehouse/checkpoints/trade_event")
  .option("mergeSchema", "true")
  .start("/lakehouse/raw/trade_event")

避坑指南： 我曾经把 failOnDataLoss 设成 true，结果 Kafka topic 被误删重建后，整个流作业直接崩溃。生产环境建议设成 false，配合告警人工介入。

4.2 批量导入：Sqoop vs Spark

实时流搞定了，但历史数据怎么办？比如过去三年的交易流水，存在 Oracle 或 MySQL 里。这时候就需要批量导入。

我早期用 Sqoop 比较多，后来慢慢转向 Spark。不是 Sqoop 不好，而是 Spark 更灵活。下面这张表对比一下：

维度 Sqoop Spark
适用场景 结构化数据，JDBC 源 结构化/半结构化，多种源
并行度控制 基于 --split-by 和 --num-mappers 基于分区数，更精细
数据转换 弱，需额外 ETL 强，内置 DataFrame API
增量导入 支持 --incremental lastmodified 需自己实现或借助 Delta 的 Change Data Feed
与 Lakehouse 集成 需先落地成文件，再加载 直接写入 Delta

我个人建议：如果源是传统关系型数据库，且数据量在 TB 级以下，Sqoop 够用；如果数据量大、需要复杂转换，或者源是 NoSQL/文件，直接用 Spark。

4.2.1 Sqoop 实战要点

用 Sqoop 从 MySQL 导数据到 HDFS，再转 Delta，我踩过一个坑：主键分布不均匀导致数据倾斜。比如用户表按自增 ID 分片，但前 1000 万条数据集中在少数 mapper 上，后面的 mapper 几乎空跑。

解决办法：用 --boundary-query 手动指定分片范围，或者改用 --split-by 选一个分布均匀的字段。

注意： Sqoop 导入时默认使用 4 个 mapper，如果数据库连接池不够，很容易把源库打挂。我建议先压测，从 2 个 mapper 开始，逐步增加。

4.2.2 Spark 批量导入

Spark 的优势在于：一次读取，多次转换，直接写入 Delta。比如从 Oracle 读数据，做类型转换、去重、加时间分区，一气呵成。

val jdbcDF = spark.read
  .format("jdbc")
  .option("url", "jdbc:oracle:thin:@//host:1521/service")
  .option("dbtable", "TRADE_HISTORY")
  .option("user", "etl_user")
  .option("password", "****")
  .option("numPartitions", "20")
  .option("partitionColumn", "TRADE_ID")
  .option("lowerBound", "1")
  .option("upperBound", "10000000")
  .load()

jdbcDF
  .withColumn("dt", to_date(col("trade_time")))
  .write
  .format("delta")
  .mode("append")
  .partitionBy("dt")
  .save("/lakehouse/raw/trade_history")

个人经验： 用 Spark 读 JDBC 时，numPartitions 不要超过数据库连接池上限。我曾经设了 50 个分区，结果 Oracle 连接池爆了，DBA 直接找我喝茶。

4.3 数据校验机制

数据进来了，怎么保证没丢、没多、没坏？这就是校验机制要干的事。我把它分成三个层次：

  • 行级校验：每条记录是否符合 schema？关键字段是否非空？
  • 批次级校验：这一批数据的总行数、总金额、哈希值是否与源端一致？
  • 端到端校验：从源到湖，延迟是否在 SLA 内？数据质量是否达标？

4.3.1 行级校验：Schema on Write

Delta Lake 默认支持 schema 校验，写入时如果字段类型不匹配，直接报错。但风控场景里，我建议再加一层软校验：把不合规的数据写到「死信队列」或单独的 _bad_records 目录，而不是直接丢弃。

// 伪代码：软校验 + 分流
val validDF = rawDF.filter(col("amount").isNotNull && col("user_id").isNotNull)
val badDF = rawDF.filter(col("amount").isNull || col("user_id").isNull)

validDF.write.format("delta").save("/lakehouse/clean/trade")
badDF.write.format("delta").save("/lakehouse/bad/trade")

4.3.2 批次级校验：Count + Checksum

批量导入时，我习惯在源端和目标端各算一次 count 和 checksum。比如 Sqoop 导入完成后，跑一个对比脚本：

-- 源端计数
SELECT COUNT(*), SUM(CAST(amount AS BIGINT)) FROM trade_history WHERE dt = '2024-01-01';

-- 目标端计数
SELECT COUNT(*), SUM(CAST(amount AS BIGINT)) FROM delta.`/lakehouse/raw/trade_history` WHERE dt = '2024-01-01';

如果对不上，就触发重跑或告警。我曾经因为网络抖动丢了一批数据，幸亏有这层校验，否则风控模型用错数据，后果不堪设想。

4.3.3 端到端校验：Data Quality Dashboard

实时流场景下，批次校验不够，因为数据是持续流入的。我建议用 Delta Live Tables (DLT) 或自定义的 Data Quality 框架，定期检查：

  • 数据延迟：当前最新数据的时间戳与系统时间差
  • 数据完整性：关键字段的空值率、重复率
  • 数据一致性：跨表关联的 key 是否匹配

一个简单的 DLT 校验示例：

CREATE STREAMING LIVE TABLE trade_validated
AS SELECT * FROM STREAM(LIVE.trade_raw)
WHERE amount > 0
  AND user_id IS NOT NULL
  AND trade_time > current_timestamp() - INTERVAL 1 HOUR

-- 自动生成数据质量指标
CONSTRAINT valid_amount EXPECT (amount > 0) ON VIOLATION DROP ROW
CONSTRAINT valid_user EXPECT (user_id IS NOT NULL) ON VIOLATION DROP ROW

重要提醒： 校验机制不是一次性工作。数据源 schema 会变，业务规则会变，校验逻辑也要跟着迭代。我每季度会 review 一次校验规则，去掉过时的，加上新的。

4.4 本章知识体系图

下面这张图把数据摄入与采集的核心链路串起来了，从源到湖，三种方式并行，校验贯穿始终。

数据摄入与采集核心链路 数据摄入与采集核心链路 业务源系统 Kafka 实时流 Sqoop 批量导入 Spark 批量导入 数据校验层 行级校验 | 批次校验 | 端到端校验 合格数据 → Delta Lake 不合格数据 → 死信队列 说明： 三种接入方式可并行使用，互不干扰。校验层是强制环节，不合格数据不进湖。 实时流推荐 Kafka + Spark Structured Streaming；批量导入根据数据量选择 Sqoop 或 Spark。

嗯，数据摄入这块，说白了就是「进得来、进得对、进得快」。Kafka 搞定实时，Sqoop/Spark 搞定批量，校验机制兜底。下一层，我们就要聊数据进了湖之后，怎么存、怎么管——那是 Lakehouse 的核心能力所在。

← 上一章 📖 返回目录 下一章 →