4. 数据采集与接入：批采集（Sqoop/DataX）、流采集（Canal/Kafka Connect）、实时接入架构设计

数据采集是批流链路的起点，决定了上游数据能否完整、及时、一致地进入金融图谱处理层。本章围绕批采集与流采集两类主流工具，以及实时接入的总体架构展开，帮助读者建立从工具选型到架构设计的完整认知。

4.1 批采集：Sqoop 与 DataX 对比

批采集适用于离线 T+1 或小时级数据同步场景，典型任务包括：从关系型数据库（MySQL/Oracle）抽取客户信息表、交易流水表到 Hive/HDFS，或从 Hive 导出计算结果到业务库。

4.1.1 DataX 批采集作业示例（MySQL → HDFS）

{
  "job": {
    "content": [
      {
        "reader": {
          "name": "mysqlreader",
          "parameter": {
            "connection": [{
              "jdbcUrl": ["jdbc:mysql://10.0.1.10:3306/finance_db"],
              "table": ["t_customer_account"]
            }],
            "username": "datax_user",
            "password": "encrypted_pwd",
            "column": ["account_id", "customer_name", "balance", "update_time"],
            "where": "update_time >= '2025-01-01 00:00:00'"
          }
        },
        "writer": {
          "name": "hdfswriter",
          "parameter": {
            "defaultFS": "hdfs://nameservice1",
            "path": "/user/finance/ods/t_customer_account",
            "fileType": "parquet",
            "writeMode": "append",
            "compress": "snappy"
          }
        }
      }
    ],
    "setting": {
      "speed": {"channel": 4},
      "errorLimit": {"record": 100}
    }
  }
}

4.2 流采集：Canal 与 Kafka Connect

流采集面向毫秒级实时接入，核心场景包括：监听 MySQL Binlog 获取交易流水、账户变更；采集 Kafka 日志进入实时计算引擎；同步 Elasticsearch 索引变更等。

4.2.1 Canal：MySQL Binlog 实时采集

Canal 模拟 MySQL Slave 节点，解析 Binlog 的 Row 模式变更事件，输出为 JSON 格式的 Insert/Update/Delete 消息。金融场景下常用于交易流水、风控规则变更的实时捕获。

# canal 实例配置示例（conf/example/instance.properties）
canal.instance.master.address=10.0.1.10:3306
canal.instance.dbUsername=canal_user
canal.instance.dbPassword=canal_pwd
canal.instance.connectionCharset=UTF-8
canal.instance.filter.regex=fins_db\\..*
# 只监听 fins_db 下所有表
canal.mq.topic=fin_binlog_topic
canal.mq.partitionsNum=6

4.2.2 Kafka Connect：多源流采集框架

Kafka Connect 是 Kafka 生态的流数据集成框架，通过 Source Connector 从外部系统拉取数据，通过 Sink Connector 写入目标系统。金融图谱中常用 Debezium（MySQL/MongoDB CDC）作为 Source，Elasticsearch Sink 作为索引同步。

4.3 实时接入架构设计

金融图谱的实时接入层需要同时处理高吞吐、低延迟、数据一致性三个核心挑战。下图展示了一套经过生产验证的批流一体接入架构。

4.3.1 架构设计要点

1. CDC 优先，轮询兜底：核心交易库使用 Canal/Debezium 监听 Binlog，实现毫秒级变更捕获；对无 Binlog 能力的系统（如部分文件接口），使用 DataX 定时轮询作为补充。
2. 消息队列解耦：所有采集数据先入 Kafka/Pulsar，避免采集端与处理端直接耦合；利用 Topic 分区机制实现并行消费，单 Topic 吞吐可达 10 万条/秒。
3. Schema 统一管理：引入 Schema Registry，确保 CDC 消息的字段变更（新增列、类型修改）能被下游 Flink 任务自动识别，避免反序列化失败。
4. 链路监控与位点回溯：在采集端记录 Binlog 位点（GTID/File+Pos），在 Kafka 端记录 Offset，支持 7 天内任意时间点的数据回溯重放。

4.4 批流采集选型决策矩阵