每天TB级数据灌进来，Binlog+日志+埋点全要——数据接入层怎么设计？
本文目录
先说结论
数据接入的核心原则：统一走Kafka，不管什么数据源。所有数据先落Kafka，再从Kafka分发到离线和实时两条链路。这是接入层最重要的设计决策。
三类数据源的接入方式：① 业务数据库（MySQL/PostgreSQL）→ CDC工具（Canal/Debezium）→ Kafka；② 服务端日志 → Flume/Filebeat → Kafka；③ 客户端埋点 → 埋点SDK → 埋点服务 → Kafka。
接入层的质量直接决定下游的质量。如果接入层丢数据、乱序、Schema不一致，下游再怎么治理都是在给上游擦屁股。
先把这段讲顺；面试官追问时，再展开下面的原理和细节。
详细讲解
数据接入是数仓的'入口'，也是最容易被忽视的一层。很多人觉得'数据接入不就是从MySQL导到Hive嘛，有什么难的'，但真正做过后才知道坑有多少——Binlog采集延迟、日志格式不统一、埋点数据质量差、多数据源的Schema漂移……我在两家公司分别搭建过数据接入层，从最初的Sqoop全量同步到后来的CDC+Kafka实时采集，今天把接入层的设计思路和踩坑全讲出来。
数据接入层整体架构
我在某短剧出海公司设计的接入层架构：
┌──────────────────────────────────────────────────────┐
│ 数据源 │
├──────────────────────────────────────────────────────┤
│ MySQL业务库 服务端日志 客户端埋点 第三方API │
│ │ │ │ │ │
│ Canal Filebeat 埋点SDK 定时拉取 │
│ │ │ │ │ │
│ └──────────────┴────────────┴─────────────┘ │
│ │ │
│ Kafka │
│ (统一消息总线) │
│ / \ │
│ Flink(实时) Kafka→MaxCompute(离线) │
└──────────────────────────────────────────────────────┘业务数据库接入：CDC采集
业务数据库的变更数据是数仓最核心的数据源之一。我在两家公司分别用过两种CDC方案：
| 维度 | Canal（阿里系） | Debezium（开源社区） |
|---|---|---|
| 支持数据库 | MySQL为主 | MySQL/PostgreSQL/MongoDB/Oracle |
| 部署方式 | 独立Server+Client | Kafka Connect插件 |
| 消息格式 | 自定义Protobuf/JSON | 标准化JSON（含before/after） |
| 社区活跃度 | 中（主要国内用） | 高（全球社区） |
| 阿里云集成 | 原生支持（DTS） | 需要自行部署 |
| 我的选择 | 短剧出海（阿里云技术栈） | 出海硬件（AWS技术栈） |
// Canal采集的Binlog消息示例
{
"database": "app_db",
"table": "user_order",
"type": "UPDATE",
"ts": 1697356800,
"data": [{
"order_id": "ORD_20231015_001",
"user_id": "u_998877",
"pay_status": "success",
"pay_amount": 9.99,
"update_time": "2023-10-15 20:00:00"
}],
"old": [{
"pay_status": "pending"
}]
}关键设计决策：Canal采集到Kafka时，一个业务表对应一个Kafka Topic（如cdc_app_db_user_order）。不要把所有表的Binlog打到同一个Topic——这样下游消费时无法按表过滤，而且不同表的消息量差异很大，会导致消费不均衡。
服务端日志接入：Filebeat → Kafka
服务端日志（API请求日志、应用日志、Nginx访问日志）通过Filebeat采集到Kafka。Filebeat相比Flume更轻量，部署在每台应用服务器上，资源占用很小。
# Filebeat配置示例：采集API请求日志
filebeat.inputs:
- type: log
paths:
- /var/log/app/api-request-*.log
json.keys_under_root: true # JSON日志直接解析
json.add_error_key: true # 解析失败时添加error字段
fields:
log_type: api_request # 添加日志类型标识
env: production # 添加环境标识
output.kafka:
hosts: ['kafka-1:9092', 'kafka-2:9092', 'kafka-3:9092']
topic: 'ods_api_request_log'
partition.round_robin:
reachable_only: true
required_acks: 1 # 至少一个Broker确认
compression: gzip # 压缩传输，节省带宽客户端埋点接入：SDK → 埋点服务 → Kafka
埋点数据是数据质量问题最多的数据源。客户端SDK采集的数据经过埋点服务（一个HTTP服务）校验后写入Kafka：
客户端埋点接入流程：
App/Web SDK 埋点服务(HTTP)
│ │
│ POST /collect │
│ {events: [...]} │
│ ─────────────────────────────→ │
│ ├─ ① 格式校验（必填字段检查）
│ ├─ ② 数据清洗（去除敏感信息）
│ ├─ ③ 服务端补充（server_time、IP解析）
│ ├─ ④ 写入Kafka（按事件类型分Topic）
│ 200 OK │
│ ←───────────────────────────── │埋点服务的关键职责：① 补充server_time（不完全信任客户端时间，用服务器时间做兜底）；② IP解析成地理位置（离线解析太慢，在接入层实时解析）；③ 格式校验（event_type和必填字段缺失的直接丢弃，并记录到异常日志）。
全量同步 vs 增量同步
业务表的接入方式选择是面试高频题：
| 维度 | 全量同步 | 增量同步（CDC） |
|---|---|---|
| 原理 | 每天定时全表SELECT导出 | 实时监听Binlog变更 |
| 延迟 | T+1（或每隔几小时） | 秒级 |
| 对源库压力 | 高（全表扫描） | 低（只读Binlog） |
| 数据完整性 | 高（每次全量保证一致） | 中（可能漏Binlog） |
| 适用场景 | 维度表、小表、首次全量加载 | 事实表、大表、实时需求 |
| 存储成本 | 高（每天一份全量） | 低（只存变更增量） |
我们的策略是：事实表（订单、事件）用CDC增量同步，维度表（渠道、国家）每天全量同步+CDC增量补充。为什么维度表要全量兜底？因为CDC可能漏数据（Canal重启丢位点），全量同步能保证维度表一定是完整的。
面试高频追问
追问一：Canal采集的Binlog有延迟怎么办？
先看延迟在哪个环节：Canal解析Binlog → 写入Kafka → 消费端处理。大多数延迟出在Canal端——MySQL主从延迟、Binlog解析速度跟不上写入速度。我们的监控方案是：Canal暴露了一个metrics端口，Prometheus采集delay指标（Canal当前解析位点和MySQL最新位点的时间差），超过30秒告警。如果是持续延迟，通常是MySQL写入量暴增（大促期间），需要考虑Canal的多实例并行解析。
追问二：数据源的Schema变了怎么办？比如MySQL加了一个字段。
这是Schema Evolution问题。上游加字段是常有的事，如果不做处理，下游Hive表缺少新字段就会丢数据。我们的方案是：① ODS层的Hive表用STRING类型存原始JSON（不用强Schema），新增字段自动包含在JSON中；② DWD层定期检查上游Schema变更（通过比对Canal消息的字段列表），发现新字段后评估是否需要加入DWD表。这样ODS永远不会丢字段，DWD按需扩展。
追问三：多个数据源的时间怎么对齐？客户端时间和服务端时间不一样。
这是出海业务的经典问题。我们的统一规则是：所有事件都存两个时间——event_time（客户端上报的事件发生时间）和server_time（服务端接收时间）。分析和报表用event_time（更准确反映用户行为），监控和排查用server_time（更稳定可靠）。如果event_time明显异常（比如跟server_time差了超过24小时），用server_time替代。
实战踩坑
在某短剧出海公司遇到过一次严重的接入层事故。某天下午3点，运营突然发现实时大屏上所有指标都变成了0。紧急排查发现Kafka集群有一个Broker磁盘满了，导致Producer写入失败，Canal的消息全部堆积在本地。但Canal本地队列也有大小限制，超过限制后Canal开始丢弃消息——相当于Binlog变更直接丢了。
紧急修复：① 清理Kafka Broker磁盘（删除过期日志段）恢复写入；② 检查Canal丢失的Binlog范围；③ 用MySQL的mysqlbinlog工具手动解析丢失时段的Binlog，转成SQL重新执行到ODS。整个修复花了4个小时。
# 紧急修复步骤
# 1. 清理Kafka磁盘：删除3天前的日志段
kafka-configs.sh --alter --entity-type topics \
--entity-name ods_binlog_order \
--add-config retention.ms=86400000 # 临时改为1天
# 2. 确认Canal丢失的Binlog范围
# Canal日志：Discard binlog event, position 1234 to 5678
# 3. 用mysqlbinlog解析丢失的Binlog
mysqlbinlog --start-position=1234 --stop-position=5678 \
/var/lib/mysql/binlog.000042 > /tmp/lost_events.sql
# 4. 监控优化：添加Kafka磁盘使用率告警（>80%预警）
# 5. 长期方案：Kafka开启log.retention.bytes限制单Topic大小教训：① Kafka磁盘必须有容量告警（我们设了80%预警、90%告警）；② Canal必须配置本地队列的持久化（而不是内存队列），避免丢消息；③ 接入层的监控优先级应该跟生产数据库同级——接入层挂了，下游全部停摆。
总结
数据接入层是数仓的'地基'，地基不稳上面全部白搭。核心设计原则：所有数据统一走Kafka、CDC和全量互补、接入层做好格式校验和时间标准化。面试时讲接入层，要能说出三类数据源（Binlog/日志/埋点）各自的采集方案、全量vs增量的选择策略、以及接入层的监控和容错机制。如果能再讲一个接入层故障的排查过程，那就更有说服力了。
数据接进来了，下一步就是清洗。下一篇聊DWD明细层的清洗到底该怎么做——很多人觉得'清洗就是去重和过滤'，其实远不止这些。
你们的数据接入用的Canal还是Debezium？有没有遇到过接入层丢数据的事故？评论区聊聊～
关注「Wood聊大数据」| 9年大数据老兵，聊技术、聊职场、聊成长