监控ETL任务状态的脚本如何编写：从原理到实战的完整指南
📖 目录导读
- 为什么需要监控ETL任务状态？ – 数据管线稳定性的核心挑战
- ETL任务状态监控的核心技术原理 – 状态码、日志、心跳与异常捕获
- 编写监控脚本前的准备工作 – 环境、依赖与状态定义
- 实战：用Python编写ETL状态监控脚本 – 完整代码解析（含日志监控、超时检测、告警通知）
- 高级技巧：集成告警与可视化 – 钉钉/Slack通知、Grafana看板
- 常见问题FAQ – 脚本死循环、历史日志干扰、多任务并发处理
- 总结与最佳实践 – 让你的ETL监控“永不下线”
为什么需要监控ETL任务状态？
Q： 不监控ETL，直接跑，出问题再手动查日志不行吗？
A： 在大数据场景下，ETL任务往往在凌晨批量执行，如果某一步骤失败（如源数据库连接超时、目标表被锁、数据倾斜导致OOM），不监控意味着数据延迟数小时，甚至产生错误数据写入下游。一份2023年的行业报告显示，超过60%的数据仓库事故是由于ETL状态未被及时发现导致的。
核心痛点：
- 任务挂起：进程未崩溃，但陷入死循环,消耗资源不推进。
- 数据不一致：增量ETL时,某批次失败导致部分数据重复或丢失。
- 性能退化：任务运行时间从30分钟暴增到2小时,不监控无法感知。
ETL任务状态监控的技术原理
1 关键状态定义
| 状态 | 含义 | 触发条件 |
|---|---|---|
| RUNNING | 正在执行 | 进程存在且未超时 |
| SUCCESS | 正常结束 | 退出码为0，且输出结果文件/表行数符合预期 |
| FAILED | 执行出错 | 退出码非0，或捕获到异常 |
| TIMEOUT | 执行超时 | 运行时间 > 预设阈值 |
| STUCK | 资源占用但无进展 | 日志文件长时间无更新（heartbeat缺失） |
2 监控手段三剑客
- 进程级监控：
ps aux | grep etl_job或系统守护进程 - 日志级监控：定时检测日志文件最后修改时间、关键词（如“ERROR”“FATAL”）
- 状态文件/心跳机制：ETL脚本每隔5秒向特定文件写入时间戳，监控脚本检查时间差
编写监控脚本前的准备工作
1 环境配置
pip install requests # 用于告警通知 pip install psutil # 进程监控
2 定义监控配置文件（config.yaml）
jobs:
- name: "daily_user_etl"
pid_file: "/var/run/etl_user.pid"
log_path: "/var/log/etl/user_etl.log"
max_run_time: 3600 # 秒
heartbeat_interval: 30 # 心跳更新间隔
alert:
webhook_url: "https://open.feishu.cn/open-apis/bot/v2/hook/xxx"
Q： 为什么不用bash脚本而用Python？
A： Python能更好地处理JSON配置、HTTP请求、日期计算、正则匹配日志中的复杂模式,Bash在字符串切分和跨平台兼容性上较弱。
实战：用Python编写ETL状态监控脚本
📝 完整代码（示例部分，可扩展）
#!/usr/bin/env python3
import os
import time
import json
import requests
import psutil
from datetime import datetime
import yaml
class ETLMonitor:
def __init__(self, config_path='config.yaml'):
with open(config_path, 'r') as f:
self.config = yaml.safe_load(f)
self.alert_url = self.config['alert']['webhook_url']
def check_job_status(self, job):
"""检查单个任务状态"""
result = {'name': job['name'], 'status': 'UNKNOWN'}
# 1. 检查进程是否存活
pid = self._read_pid(job['pid_file'])
if pid and psutil.pid_exists(pid):
result['process_alive'] = True
# 2. 检查运行时间是否超时
proc = psutil.Process(pid)
run_duration = time.time() - proc.create_time()
if run_duration > job['max_run_time']:
result['status'] = 'TIMEOUT'
result['detail'] = f"运行时间 {run_duration:.0f}s 超过阈值 {job['max_run_time']}s"
else:
# 3. 检查心跳（日志更新时间）
last_modified = os.path.getmtime(job['log_path'])
if time.time() - last_modified > job['heartbeat_interval'] * 3:
result['status'] = 'STUCK'
result['detail'] = "日志最后更新超过心跳间隔3倍"
else:
# 4. 检查日志是否有错误
if self._check_log_error(job['log_path']):
result['status'] = 'FAILED'
result['detail'] = "日志包含ERROR/FATAL关键字"
else:
result['status'] = 'RUNNING'
result['detail'] = "正常运行中"
else:
# 进程不存在：检测是否有退出标志
if os.path.exists(job['pid_file']):
os.remove(job['pid_file'])
# 检查最近一次完成记录
result['status'] = 'FINISHED_OR_CRASHED'
result['detail'] = "进程不存在，请检查日志确认完成还是崩溃"
return result
def _read_pid(self, pid_path):
try:
with open(pid_path, 'r') as f:
return int(f.read().strip())
except:
return None
def _check_log_error(self, log_path, keywords=['ERROR', 'FATAL', 'ORA-']):
"""检测日志中最近200行是否含错误关键字"""
try:
with open(log_path, 'rb') as f:
# 读取末尾8KB
f.seek(-min(8192, os.path.getsize(log_path)), 2)
tail = f.read().decode('utf-8', errors='ignore')
for kw in keywords:
if kw in tail:
return True
return False
except:
return False
def send_alert(self, job_result):
"""发送告警到飞书/钉钉"""
if job_result['status'] in ['TIMEOUT', 'FAILED', 'STUCK']:
msg = {
"msg_type": "text",
"content": {
"text": f"🚨 ETL告警\n任务：{job_result['name']}\n状态：{job_result['status']}\n详情：{job_result['detail']}\n时间：{datetime.now()}"
}
}
requests.post(self.alert_url, json=msg, timeout=5)
def run_cycle(self):
"""主循环"""
while True:
for job in self.config['jobs']:
status = self.check_job_status(job)
print(f"[{datetime.now()}] {job['name']}: {status['status']}")
if status['status'] != 'RUNNING':
self.send_alert(status)
time.sleep(60) # 每分钟检查一次
if __name__ == "__main__":
monitor = ETLMonitor()
monitor.run_cycle()
🔧 使用说明
- 将上述代码保存为
etl_monitor.py - 创建
config.yaml填入具体任务参数 - 执行
nohup python3 etl_monitor.py &后台运行 - 日志会自动输出实时状态
高级技巧：集成告警与可视化
1 多通道告警
除了HTTP通知,还可发送：
- 钉钉：改用
https://oapi.dingtalk.com/robot/send?access_token=xxx - 企业微信：
https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx - 邮件：使用
smtplib发送失败摘要
2 将指标写入Prometheus
在脚本中增加时间戳记录：
from prometheus_client import Gauge, start_http_server
etl_status = Gauge('etl_job_status', '1=ok, 0=fail', ['job_name'])
暴露/metrics端口,配合Grafana绘制实时仪表盘。
Q： 监控脚本本身崩溃了怎么办？
A： 使用systemd管理监控脚本，设置Restart=always，或者使用进程守护工具如supervisord。
常见问题FAQ
Q1：脚本发现进程存在但日志长时间不更新，怎么区分是真的阻塞还是日志缓冲？
A： 配合心跳文件机制，让ETL脚本每30秒向独立的心跳文件写入时间戳，监控脚本仅依赖心跳文件,避免日志刷盘延迟干扰。
Q2：多个ETL任务共用一个日志文件怎么办？
A： 在配置中为每个任务指定独立日志文件，如果无法更改，使用正则匹配任务标识（如 [daily_user] 前缀）。
Q3：如何避免监控脚本在任务正常结束时误报“进程不存在”？
A： 在ETL脚本退出前，创建一个.done标记文件，监控脚本先检测done文件,如果存在则跳过告警并清理该文件。
总结与最佳实践
编写ETL监控脚本的核心是建立多维度检测机制：
- 进程存活 + 时间超限（最基础）
- 心跳超时 + 日志错误（检测阻塞）
- 数据量变化（可选：比如目标表行数是否增加）
最佳实践清单：
- ✅ 监控脚本本身使用
systemd或supervisor保持常驻 - ✅ 每轮检查间隔不超过任务运行时间的1/10（如任务1小时，检查间隔≤6分钟）
- ✅ 告警消息包含任务名、上下文（如最后100行日志）、直接操作建议
- ✅ 设置“静默期”：任务启动后前60秒不告警，避免启动时临时文件未就绪的误报
- ✅ 保留历史的监控数据（写入数据库），用于分析任务执行趋势
不要过于依赖“进程存在”这一指标——许多ETL框架（如Apache Airflow）有外部任务调度器，进程可能已被调度器回收，但任务实际失败，最佳方案是将监控脚本与ETL框架的状态API结合（如读取Airflow的dag_run.status）。
通过本指南，您应该能够独立编写一个生产可用的ETL状态监控脚本，实际部署时,根据您的ETL框架和告警偏好调整细节即可。
