4. 数据源接入设计：四种主流接入方式

数据源接入，说白了就是解决「数据从哪来」的问题。我做了这么多年数据服务，见过太多团队在接入层就翻车——要么丢数据，要么延迟爆炸，要么运维想骂人。

这一章，我把最常见的四种接入方式掰开揉碎讲清楚：MySQL CDC、Kafka Topic、文件系统、API 回调。每种我都会给出实战中的选型建议和避坑经验。

4.1 MySQL CDC 接入：从 Binlog 到数据管道

MySQL CDC（Change Data Capture）是目前实时数据管道中最常用的接入方式。它的原理很简单：解析 MySQL 的 Binlog，把每一行变更变成一条消息。

我个人习惯用 Debezium + Kafka Connect 来做 CDC 接入。为什么？因为 Debezium 对 MySQL 的兼容性最好，而且能自动处理 Schema 变更。

下面是一个典型的 Flink CDC 接入代码片段：

// 创建 MySQL CDC Source
DataStreamSource<String> source = MySQLSource.<String>builder()
    .hostname("192.168.1.100")
    .port(3306)
    .databaseList("order_db")
    .tableList("order_db.orders")
    .username("cdc_user")
    .password("cdc_pass")
    .deserializer(new JsonDebeziumDeserializationSchema())
    .startupOptions(StartupOptions.initial())
    .build();

// 接入流批一体管道
env.addSource(source)
   .map(new OrderTransformFunction())
   .addSink(new KafkaSink<>(...));

关于 startupOptions，我建议这样选：

启动模式	适用场景	注意事项
initial	首次接入，需要全量+增量	数据量大时可能超时
latest	只关心后续变更	会丢失历史数据
timestamp	指定时间点开始消费	需要确认 Binlog 保留时长
specificOffset	故障恢复，精确续接	需要记录 offset 信息

嗯，这里要注意：CDC 接入最大的坑是 大事务。我曾经遇到一个团队，一次批量更新 100 万行数据，Binlog 直接撑爆了 Kafka 消息大小限制。解决方案是调整 Debezium 的 max.batch.size 和 max.queue.size，或者在 Flink 端做窗口切分。

4.2 Kafka Topic 接入：消息队列的流式消费

Kafka Topic 接入，说白了就是订阅一个消息队列。这是流批一体中最「原生」的接入方式——数据已经在流里了，你只需要消费它。

我见过很多团队直接把 Kafka 当数据库用，这是不对的。Kafka 的定位是 消息管道，不是存储系统。它的数据有保留期限，默认 7 天。

Kafka 接入的核心参数配置：

// Flink Kafka Consumer 配置
Properties props = new Properties();
props.setProperty("bootstrap.servers", "kafka-1:9092,kafka-2:9092");
props.setProperty("group.id", "api-service-group");
props.setProperty("auto.offset.reset", "earliest");
props.setProperty("enable.auto.commit", "false");
props.setProperty("max.poll.records", "500");

DataStreamSource<String> stream = env
    .addSource(new FlinkKafkaConsumer<>("order-topic", 
        new SimpleStringSchema(), props));

Kafka Topic 接入时，我建议关注三个维度：

• 分区数：分区数决定了并行度上限。一般建议分区数 = 消费端并行度 × 2~3
• 消息大小：默认 1MB，如果单条消息超过这个值，需要调整 broker 和 consumer 两端的配置
• 序列化格式：JSON 方便调试但体积大，Avro/Protobuf 性能更好。我个人偏好 Avro + Schema Registry

4.3 文件系统接入：批量数据的优雅处理

文件系统接入，听起来很传统，但在流批一体架构中它依然重要。很多业务数据还是以文件形式交付的——比如第三方对账单、日志导出、历史数据迁移。

我常用的文件格式优先级：Parquet > ORC > Avro > CSV > JSON。Parquet 的列式存储对分析型查询特别友好，压缩比也高。

文件系统接入的典型流程：

// Flink 读取 HDFS 上的 Parquet 文件
DataStream<RowData> stream = env
    .createInput(FileSource
        .forRecordStreamFormat(
            new ParquetColumnarRowInputFormat(...),
            new Path("hdfs://data-lake/orders/dt=2024-01-01/"))
        .monitorContinuously(Duration.ofHours(1))
        .build());

文件接入的两种模式：

模式	说明	适用场景
批量读取	一次性读取所有文件，处理完退出	历史数据迁移、T+1 报表
持续监控	监控目录，新文件到达后自动触发	准实时文件导入、日志采集

另外，文件系统的 Schema 推断是个大坑。CSV 文件没有 Schema 信息，全靠猜。我建议在文件接入层强制要求提供 Schema 文件，或者在元数据中心注册表结构。

4.4 API 回调接入：被动接收的优雅方案

API 回调接入，也叫 Webhook 接入。它和其他三种方式最大的区别是：数据不是你去拉的，是别人推给你的。

这种模式在以下场景特别常见：

• 支付回调：支付宝/支付成功后通知你的系统
• SaaS 事件：Salesforce、Shopify 等平台推送业务事件
• 内部系统联动：A 系统处理完，回调 B 系统

API 回调接入的设计要点：

// 一个简单的 Webhook 接收服务（Spring Boot）
@PostMapping("/webhook/order")
public ResponseEntity<String> receiveOrderWebhook(
        @RequestBody OrderEvent event,
        @RequestHeader("X-Signature") String signature) {
    
    // 1. 验签（必须！）
    if (!verifySignature(event, signature)) {
        return ResponseEntity.status(401).body("invalid signature");
    }
    
    // 2. 写入消息队列，异步处理
    kafkaTemplate.send("webhook-orders", event);
    
    // 3. 立即返回 200，避免上游超时重试
    return ResponseEntity.ok("received");
}

我建议的 API 回调接入架构：

• 接入层：Nginx/网关做限流和 IP 白名单
• 验签层：HMAC-SHA256 或 RSA 签名验证
• 缓冲层：写入 Kafka/RocketMQ，削峰填谷
• 处理层：流批一体引擎消费消息，做业务处理

4.5 四种接入方式的选型对比

最后，我把四种接入方式放在一起做个对比。这样你在实际项目中可以快速决策：

维度	MySQL CDC	Kafka Topic	文件系统	API 回调
实时性	秒级	毫秒级	分钟~小时级	秒级
数据完整性	高（Binlog 保证）	中（取决于生产者）	高（文件完整后读取）	低（可能丢消息）
运维复杂度	高（Binlog 管理）	中（Topic 管理）	低（文件监控）	中（接口维护）
适用场景	数据库变更捕获	消息解耦	批量导入/导出	外部系统回调
流批一体适配	⭐⭐⭐⭐	⭐⭐⭐⭐⭐	⭐⭐⭐	⭐⭐⭐⭐