数据管道与仪表盘

手动整理数据做报表是最消耗时间的办公活动之一。自动化数据管道 + 仪表盘可以让报表"自己跑"。

自动化数据管道架构

轻量数据管道

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import csv
import io
class StepStatus(Enum):
PENDING = "pending"
RUNNING = "running"
SUCCESS = "success"
FAILED = "failed"
@dataclass
class PipelineStep:
name: str
transform_fn: str   # 转换函数名（描述用）
status: StepStatus = StepStatus.PENDING
row_count_in: int = 0
row_count_out: int = 0
error: str = ""
@dataclass
class DataPipeline:
"""轻量 ETL 数据管道"""
name: str
steps: list[PipelineStep] = field(default_factory=list)
created_at: str = field(
default_factory=lambda: datetime.now().strftime("%Y-%m-%d %H:%M")
)
def add_step(self, name: str, transform_fn: str) -> None:
self.steps.append(PipelineStep(name=name, transform_fn=transform_fn))
def run(self, data: list[dict]) -> list[dict]:
"""依次执行管道步骤"""
current = data
for step in self.steps:
step.status = StepStatus.RUNNING
step.row_count_in = len(current)
try:
current = self._apply_step(step, current)
step.row_count_out = len(current)
step.status = StepStatus.SUCCESS
except Exception as e:
step.error = str(e)
step.status = StepStatus.FAILED
break
return current
def _apply_step(
self, step: PipelineStep, data: list[dict]
) -> list[dict]:
"""应用单步转换"""
fn_name = step.transform_fn
if fn_name == "remove_nulls":
return [
row for row in data
if all(v is not None for v in row.values())
]
elif fn_name == "deduplicate":
seen = set()
unique = []
for row in data:
key = tuple(sorted(row.items()))
if key not in seen:
seen.add(key)
unique.append(row)
return unique
elif fn_name == "normalize_numbers":
for row in data:
for k, v in row.items():
if isinstance(v, str) and v.replace(".", "").isdigit():
row[k] = float(v)
return data
else:
return data
def summary(self) -> str:
lines = [f"管道: {self.name}"]
for i, s in enumerate(self.steps, 1):
lines.append(
f"  步骤{i}: {s.name} [{s.status.value}] "
f"{s.row_count_in}→{s.row_count_out} rows"
)
return "\n".join(lines)
# 使用示例
pipeline = DataPipeline(name="月度销售清洗")
pipeline.add_step("去空值", "remove_nulls")
pipeline.add_step("去重", "deduplicate")
pipeline.add_step("数字标准化", "normalize_numbers")
raw_data = [
{"product": "A", "sales": "100.5"},
{"product": "B", "sales": None},
{"product": "A", "sales": "100.5"},
{"product": "C", "sales": "200"},
]
clean = pipeline.run(raw_data)
print(pipeline.summary())

仪表盘数据聚合

from dataclasses import dataclass
@dataclass
class DashboardMetric:
label: str
value: float
unit: str
trend: str = ""   # ↑ / ↓ / →
class DashboardBuilder:
"""仪表盘数据构建器"""
def aggregate_sales(self, records: list[dict]) -> list[DashboardMetric]:
"""从销售数据生成仪表盘指标"""
if not records:
return []
total = sum(float(r.get("amount", 0)) for r in records)
count = len(records)
avg = total / count if count else 0
# 按月分组统计
monthly: dict[str, float] = {}
for r in records:
month = r.get("date", "")[:7]
monthly[month] = monthly.get(month, 0) + float(r.get("amount", 0))
months = sorted(monthly.keys())
if len(months) >= 2:
last = monthly[months[-1]]
prev = monthly[months[-2]]
trend = "↑" if last > prev else "↓" if last < prev else "→"
else:
trend = "→"
return [
DashboardMetric("总销售额", total, "元", trend),
DashboardMetric("订单数", count, "笔"),
DashboardMetric("平均客单价", round(avg, 2), "元"),
DashboardMetric("月份数", len(months), "月"),
]
def to_table(self, metrics: list[DashboardMetric]) -> str:
"""输出为 Markdown 表格"""
lines = ["| 指标 | 数值 | 单位 | 趋势 |", "|------|------|------|------|"]
for m in metrics:
lines.append(f"| {m.label} | {m.value:,.2f} | {m.unit} | {m.trend} |")
return "\n".join(lines)

报表工具对比

自动化调度模式

本章小结

• 管道化思维——采集 → 清洗 → 存储 → 可视化，每步独立可测试
• 内置转换器——去空值、去重、数字标准化覆盖 80% 常见需求
• 指标聚合——自动计算总额、均值、趋势方向
• 工具选型——轻量用 Sheets/Excel，团队用 Metabase，企业用 Power BI
• 跳过空轮——数据无变化时不重复生成，节省资源

下一章：自动化监控与异常处理