我用 Python 搭了一套实时数据监控面板从数据采集到可视化大屏附完整代码适合需要监控业务数据网站流量、销售数据、服务器状态等的开发者和运营。本文用 Python Streamlit 搭了一套实时监控面板从数据采集到可视化展示附完整代码。背景为什么需要实时监控很多业务数据分散在不同系统里——网站流量在 Google Analytics销售数据在 Excel服务器状态在运维后台。每天要打开 3-5 个系统才能了解全貌。实时监控面板的价值把所有关键数据集中在一个页面实时更新一眼看清。技术选型方案优点缺点推荐Streamlit最简单纯 Python自定义样式有限⭐⭐⭐⭐⭐Dash (Plotly)图表强大交互性好学习曲线稍陡⭐⭐⭐⭐Grafana专业监控插件丰富需要额外部署⭐⭐⭐⭐前端自研完全可控开发成本高⭐⭐我选 Streamlit理由纯 Python不需要写前端代码10 分钟能搭出一个能用的面板。模块 1数据采集从多个来源采集数据。importpandasaspdimportrequestsimportsqlite3fromdatetimeimportdatetime,timedeltaclassDataCollector:多来源数据采集器def__init__(self):self.dbsqlite3.connect(monitor.db)self._init_db()def_init_db):初始化数据库表self.db.execute( CREATE TABLE IF NOT EXISTS metrics ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP, source TEXT, metric_name TEXT, metric_value REAL, metadata TEXT ) )self.db.commit()defcollect_from_api(self,url,source_name):从 API 采集数据try:resprequests.get(url,timeout10)dataresp.json()forkey,valueindata.items():ifisinstance(value,(int,float)):self.db.execute(INSERT INTO metrics (source, metric_name, metric_value) VALUES (?, ?, ?),(source_name,key,value))self.db.commit()print(f✅{source_name}: 采集成功)exceptExceptionase:print(f❌{source_name}:{e})defcollect_from_csv(self,file_path,source_name):从 CSV 采集数据dfpd.read_csv(file_path)forcolindf.select_dtypes(includenumber).columns:avg_valdf[col].mean()self.db.execute(INSERT INTO metrics (source, metric_name, metric_value) VALUES (?, ?, ?),(source_name,col,avg_val))self.db.commit()print(f✅{source_name}: 采集成功)defget_recent(self,sourceNone,hours24):获取最近 N 小时的数据sincedatetime.now()-timedelta(hourshours)querySELECT * FROM metrics WHERE timestamp ?params[since]ifsource:query AND source ?params.append(source)returnpd.read_sql(query,self.db,paramsparams)模块 2Streamlit 面板importstreamlitasstimportplotly.expressaspximportplotly.graph_objectsasgo st.set_page_config(page_title数据监控面板,layoutwide)st.title( 实时数据监控面板)# 侧边栏数据源选择st.sidebar.header(设置)refresh_intervalst.sidebar.slider(刷新间隔秒,10,300,60)selected_sourcesst.sidebar.multiselect(数据源,[网站流量,销售数据,服务器状态,API 调用],default[网站流量,销售数据])# 数据采集collectorDataCollector()# 顶部关键指标卡片col1,col2,col3,col4st.columns(4)withcol1:st.metric(今日访问,12,345,8.5%)withcol2:st.metric(今日收入,¥45,678,12.3%)withcol3:st.metric(活跃用户,1,234,-2.1%)withcol4:st.metric(API 调用,56,789,15.6%)# 图表区域st.markdown(---)# 图表 1趋势图st.subheader( 趋势分析)df_trendcollector.get_recent(hours168)# 最近 7 天ifnotdf_trend.empty:figpx.line(df_trend,xtimestamp,ymetric_value,colorsource,title7 天趋势)st.plotly_chart(fig,use_container_widthTrue)# 图表 2分布图col_left,col_rightst.columns(2)withcol_left:st.subheader( 数据分布)fig2px.pie(df_trend,namessource,valuesmetric_value,title各数据源占比)st.plotly_chart(fig2,use_container_widthTrue)withcol_right:st.subheader( 实时指标)fig3go.Figure(go.Indicator(modegaugenumberdelta,value85,title{text:系统负载},delta{reference:80},gauge{axis:{range:[0,100]},bar:{color:darkblue},steps:[{range:[0,50],color:lightgreen},{range:[50,80],color:yellow},{range:[80,100],color:red}]}))st.plotly_chart(fig3,use_container_widthTrue)# 图表 3数据表格st.subheader( 最近数据)st.dataframe(df_trend.tail(20),use_container_widthTrue)# 自动刷新st.markdown(f script setTimeout(function(){{ window.location.reload(); }},{refresh_interval*1000}); /script ,unsafe_allow_htmlTrue)模块 3告警系统当数据异常时自动通知。importsmtplibfromemail.mime.textimportMIMETextclassAlertSystem:数据告警系统def__init__(self,rulesNone):self.rulesrulesor[]defadd_rule(self,metric_name,condition,threshold,message):添加告警规则self.rules.append({metric:metric_name,condition:condition,# gt, lt, eqthreshold:threshold,message:message})defcheck(self,metrics):检查是否触发告警alerts[]forruleinself.rules:valuemetrics.get(rule[metric])ifvalueisNone:continuetriggeredFalseifrule[condition]gtandvaluerule[threshold]:triggeredTrueelifrule[condition]ltandvaluerule[threshold]:triggeredTrueelifrule[condition]eqandvaluerule[threshold]:triggeredTrueiftriggered:alerts.append({metric:rule[metric],value:value,threshold:rule[threshold],message:rule[message]})returnalertsdefsend_email_alert(self,alerts,to_email):发送邮件告警ifnotalerts:returnbody⚠️ 数据告警\n\nforalertinalerts:bodyf-{alert[message]}: 当前值{alert[value]}, 阈值{alert[threshold]}\nmsgMIMEText(body)msg[Subject]数据监控告警msg[From]monitorexample.commsg[To]to_email# 发送邮件# with smtplib.SMTP(smtp.example.com, 587) as server:# server.login(user, password)# server.send_message(msg)# 使用alert_systemAlertSystem()alert_system.add_rule(error_rate,gt,5,错误率超过 5%)alert_system.add_rule(response_time,gt,2000,响应时间超过 2 秒)alert_system.add_rule(cpu_usage,gt,90,CPU 使用率超过 90%)alertsalert_system.check({error_rate:8,response_time:1500,cpu_usage:95})# 触发错误率超过 5%、CPU 使用率超过 90%模块 4定时数据采集importscheduleimporttimedefscheduled_collect():定时采集数据collectorDataCollector()# 采集各数据源collector.collect_from_api(https://api.example.com/stats,网站流量)collector.collect_from_api(https://api.example.com/sales,销售数据)# 检查告警recentcollector.get_recent(hours1)ifnotrecent.empty:latestrecent.iloc[-1]alertsalert_system.check({latest[metric_name]:latest[metric_value]})ifalerts:alert_system.send_email_alert(alerts,adminexample.com)# 每 5 分钟采集一次schedule.every(5).minutes.do(scheduled_collect)whileTrue:schedule.run_pending()time.sleep(1)启动面板# 安装依赖pipinstallstreamlit plotly pandas requests# 启动面板streamlit run dashboard.py# 浏览器打开 http://localhost:8501踩坑记录坑 1Streamlit 每次交互都重新执行整个脚本症状每次点击按钮所有数据采集都重新跑一遍很慢。原因Streamlit 的执行模型是每次交互都从头执行。解决用st.cache_data缓存数据采集结果。st.cache_data(ttl60)# 缓存 60 秒defget_data():returncollector.get_recent(hours24)坑 2图表太多导致页面加载慢症状页面上有 10 个图表加载要 10 秒。原因每个图表都是一次独立的渲染请求。解决用st.tabs分标签页只渲染当前可见的图表。坑 3数据量太大内存爆了症状采集了 100 万条数据内存占用 2GB。原因所有数据都加载到内存。解决只加载最近 N 小时的数据用 SQL 在数据库端过滤。坑 4告警规则太敏感症状CPU 使用率偶尔飙到 91%触发告警但实际上没问题。原因瞬时值触发告警应该看平均值。解决告警规则改为连续 3 次超过阈值才触发。坑 5多人同时访问面板数据不一致症状A 看到的数据和 B 看到的不一样。原因每个人访问时都重新采集数据时间点不同。解决数据采集和展示分离——定时任务采集存数据库面板只读数据库。总结3 条核心经验Streamlit 是最快的可视化方案。纯 Python不需要写前端10 分钟出一个能用的面板。数据采集和展示要分离。定时任务采集存数据库面板只读数据库。不要在面板里直接采集。告警规则要设防抖。不要看瞬时值看连续 N 次的平均值避免误报。你有搭建过数据监控面板吗用什么技术栈评论区交流。