4、数据源层监控：数据库Binlog解析延迟、API接口响应时间、文件到达时间监控（SFTP/HDFS）
数据源层，说白了就是咱们整个数据管道的水龙头。水龙头出水慢了，或者干脆不出水，下游再折腾也没用。这一章，我重点聊聊三个最常见的源头监控场景：数据库的 Binlog 解析、API 接口响应、以及文件到达（SFTP/HDFS）。
嗯，这里要注意，数据源层的监控核心就一句话：源头数据必须在约定时间内，以约定格式，到达约定位置。做不到，新鲜度就是空谈。
4.1 数据库 Binlog 解析延迟监控
先说说 Binlog。很多实时数仓都靠解析 MySQL 的 Binlog 来捕获变更。我见过不少团队，Binlog 解析程序跑得好好的，突然某天延迟飙到几十分钟，一问，是 DBA 那边做了主从切换，Binlog 文件名变了，解析程序没适配。
监控 Binlog 解析延迟，核心是两个指标：
- 当前解析位点与数据库最新位点的时间差
- 解析程序自身的心跳
具体怎么做？我个人习惯在解析程序里埋点，定期上报当前解析到的 Binlog 文件名和位置（Position），然后跟数据库主库的当前位点做对比。
核心公式：
解析延迟 = 数据库当前事务时间 - 最近解析事务时间
单位：秒。阈值建议：正常业务 < 5s，告警阈值 30s，严重告警 120s。
代码层面，我通常会写一个简单的上报逻辑，比如用 Java 的 Canal 客户端或者 Python 的 pymysql-replication：
# 伪代码：Binlog解析延迟上报
def report_binlog_lag(conn, log_file, log_pos):
# 获取数据库当前GTID或事务时间
cursor = conn.cursor()
cursor.execute("SELECT NOW()")
db_time = cursor.fetchone()[0]
# 获取当前解析到的Binlog时间（从事件中提取）
parsed_time = get_parsed_event_time(log_file, log_pos)
lag_seconds = (db_time - parsed_time).total_seconds()
# 上报到监控系统（Prometheus / InfluxDB）
metrics_gauge('binlog_parse_lag_seconds', lag_seconds,
tags={'host': socket.gethostname(), 'binlog_file': log_file})
if lag_seconds > 120:
alert_manager.send_alert('Binlog解析延迟严重', lag_seconds)
我曾经踩过一个坑： 只监控了位点差，没监控解析程序是否还活着。有一次程序 OOM 了，Binlog 位点停留在 10 分钟前，监控显示延迟 10 分钟，没触发告警（阈值设的 30 分钟）。结果业务方发现数据 1 小时没更新，才排查到程序挂了。后来我加了一个心跳指标：如果解析程序连续 3 个上报周期（比如 30 秒）没有上报任何位点，直接触发 P0 告警。
4.2 API 接口响应时间监控
API 接口是另一个常见的数据源。特别是对接外部系统，比如支付回调、用户行为上报。接口响应慢，直接拖慢整个数据链路的时效。
监控 API 响应时间，我建议分层：
- 客户端视角： 从发起请求到收到完整响应的时间（包括网络延迟）
- 服务端视角： 接口实际处理时间（排除网络）
- 超时与错误率： 响应时间超过阈值的请求占比
你想想看，如果客户端监控到平均响应时间 200ms，但服务端监控显示平均处理时间只有 50ms，那问题大概率出在网络或者负载均衡上。
| 指标 | 正常范围 | 告警阈值 | 严重阈值 |
|---|---|---|---|
| P50 响应时间 | < 200ms | 500ms | 1000ms |
| P99 响应时间 | < 500ms | 2000ms | 5000ms |
| 超时率（>5s） | < 0.1% | 1% | 5% |
| 错误率（4xx/5xx） | < 0.5% | 2% | 10% |
在实际项目中，我习惯用 Prometheus 的 Histogram 来采集 API 响应时间。配合 Grafana 的 Heatmap，能很直观地看到响应时间的分布变化。
# Prometheus Histogram 示例（Java Micrometer）
@Timed(value = "api.response.time",
histogram = true,
extraTags = {"endpoint", "/api/v1/orders"})
public Response getOrders() {
// 业务逻辑
}
一个小技巧： 监控 API 响应时间时，别忘了记录请求体大小。我遇到过接口本身很快，但请求体太大（比如 10MB 的 JSON），导致序列化/反序列化耗时飙升。加上请求体大小标签，能快速定位这类问题。
4.3 文件到达时间监控（SFTP / HDFS）
文件型数据源，比如业务方每天凌晨通过 SFTP 上传的日志文件，或者 HDFS 上的离线分区数据。这类监控的核心是：文件是否在预期时间内到达。
说白了，就是定一个“最晚到达时间”，到了这个点文件没来，就告警。
我建议把文件到达监控拆成三个维度：
- 文件存在性： 指定路径下，指定文件名模式的文件是否存在
- 文件完整性： 文件大小是否稳定（不再增长），或者是否有 .done / .flag 标记文件
- 文件时效性： 文件的最后修改时间是否在预期窗口内
举个例子，业务方承诺每天凌晨 3 点前上传前一天的交易日志到 SFTP。那么监控脚本可以这样写：
# 伪代码：SFTP文件到达监控
def check_sftp_file_arrival(sftp_host, sftp_path, file_pattern, deadline_time):
# 连接SFTP
transport = paramiko.Transport((sftp_host, 22))
transport.connect(username='user', password='pass')
sftp = paramiko.SFTPClient.from_transport(transport)
# 列出匹配的文件
files = sftp.listdir_attr(sftp_path)
matched = [f for f in files if fnmatch.fnmatch(f.filename, file_pattern)]
if not matched:
# 没有文件，触发告警
alert_manager.send_alert(f'文件未到达: {file_pattern}',
f'截止时间: {deadline_time}')
return
# 检查最新文件的修改时间
latest_file = max(matched, key=lambda f: f.st_mtime)
file_mtime = datetime.fromtimestamp(latest_file.st_mtime)
if file_mtime > deadline_time:
# 文件在截止时间之后才修改，说明延迟
delay_minutes = (file_mtime - deadline_time).total_seconds() / 60
alert_manager.send_alert(f'文件延迟到达: {latest_file.filename}',
f'延迟: {delay_minutes:.1f} 分钟')
# 检查文件大小是否稳定（假设10分钟没有变化视为写完）
if latest_file.st_size < 1024: # 小于1KB，可能是空文件
alert_manager.send_alert(f'文件过小: {latest_file.filename}',
f'大小: {latest_file.st_size} bytes')
sftp.close()
我曾经遇到过一个经典问题： 监控脚本检查文件到达时间，用的是文件的修改时间（mtime）。结果有一次 SFTP 服务器时间不准，比实际时间慢了 2 小时，导致监控一直误报。后来我改成在监控脚本里先做一次 NTP 时间同步校验，或者直接取文件名中的时间戳（比如 log_20240520_030000.csv），而不是依赖文件系统时间。
4.4 三种数据源的监控对比
为了方便你快速对比，我把这三种数据源的核心监控要点整理成了一张表：
| 数据源类型 | 核心监控指标 | 常用工具/方法 | 典型告警阈值 |
|---|---|---|---|
| MySQL Binlog | 解析延迟（秒）、心跳、位点 | Canal / Maxwell / Debezium + JMX | 延迟 > 30s 告警，> 120s 严重 |
| API 接口 | P50/P99 响应时间、超时率、错误率 | Prometheus + Grafana / SkyWalking | P99 > 2s 告警，> 5s 严重 |
| 文件（SFTP/HDFS） | 文件存在性、修改时间、大小稳定性 | Paramiko / HDFS CLI + Cron + 告警 | 超过约定时间 30 分钟未到达告警 |
4.5 数据源层监控的整体架构
最后，我画了一张图，把这三类监控的采集、上报、告警链路串起来。你一看就明白了。
这张图里，我特意把采集层放在中间。不管你用 Canal Exporter、Prometheus Blackbox Exporter，还是自己写的 Cron 脚本，最终都要把数据汇聚到统一的存储里。我个人习惯用 Prometheus，因为它生态好，Grafana 支持也最成熟。
好了，数据源层的监控就聊到这儿。记住，源头不稳，后面全是白搭。下一层，我们该看看数据在管道里传输时怎么监控了。