1. 构建时序数据管道

任务说明

构建一个可扩展的时序数据摄取管道，支持从 CSV 文件与 REST API 两种数据源摄取数据，统一输出为带时间戳校验的标准化记录流，并确保数据完整性与时间戳正确性。管道核心逻辑独立于数据源，新增数据源时只需实现统一接口。

环境依赖

• Python 3.9+
• 依赖库：requests（仅 API 源需要）、标准库 csv、json、datetime

核心实现

import csv
import json
from datetime import datetime, timezone
from typing import Dict, Iterable, List, Union
from abc import ABC, abstractmethod

class TimeSeriesRecord:
    """标准化时序记录，校验时间戳格式与完整性"""
    REQUIRED_FIELDS = {"timestamp", "value"}
    
    def __init__(self, timestamp: str, value: float, tags: Dict[str, str] = None):
        # 时间戳必须为 ISO 8601 格式且带时区
        try:
            parsed = datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
        except ValueError:
            raise ValueError(f"非法时间戳格式: {timestamp}")
        if parsed.tzinfo is None:
            raise ValueError(f"时间戳缺少时区信息: {timestamp}")
        self.timestamp = parsed.astimezone(timezone.utc).isoformat()
        self.value = float(value)
        self.tags = tags or {}

    def to_dict(self) -> Dict:
        return {"timestamp": self.timestamp, "value": self.value, "tags": self.tags}

class DataSource(ABC):
    """数据源统一接口：返回原始字典迭代器"""
    @abstractmethod
    def read_raw(self) -> Iterable[Dict]:
        pass

class CSVSource(DataSource):
    def __init__(self, filepath: str, timestamp_col: str = "timestamp", 
                 value_col: str = "value", tag_cols: List[str] = None):
        self.filepath = filepath
        self.timestamp_col = timestamp_col
        self.value_col = value_col
        self.tag_cols = tag_cols or []

    def read_raw(self) -> Iterable[Dict]:
        with open(self.filepath, newline='', encoding='utf-8') as f:
            reader = csv.DictReader(f)
            for row in reader:
                yield {
                    "timestamp": row[self.timestamp_col],
                    "value": row[self.value_col],
                    "tags": {c: row[c] for c in self.tag_cols if c in row}
                }

class APISource(DataSource):
    def __init__(self, endpoint: str, params: Dict = None):
        self.endpoint = endpoint
        self.params = params or {}

    def read_raw(self) -> Iterable[Dict]:
        import requests
        resp = requests.get(self.endpoint, params=self.params, timeout=10)
        resp.raise_for_status()
        data = resp.json()
        # 支持两种常见 API 返回结构：数组或 {"data": [...]}
        items = data if isinstance(data, list) else data.get("data", [])
        for item in items:
            yield {
                "timestamp": item["timestamp"],
                "value": item["value"],
                "tags": item.get("tags", {})
            }

class IngestionPipeline:
    """摄取管道：读取原始数据 → 校验转换 → 输出标准化记录"""
    def __init__(self, source: DataSource, batch_size: int = 1000):
        self.source = source
        self.batch_size = batch_size
        self._stats = {"total": 0, "invalid": 0}

    def run(self) -> List[Dict]:
        """执行摄取，返回标准化记录列表；无效记录跳过并计数"""
        results = []
        for raw in self.source.read_raw():
            try:
                record = TimeSeriesRecord(
                    timestamp=raw["timestamp"],
                    value=raw["value"],
                    tags=raw.get("tags")
                )
                results.append(record.to_dict())
                self._stats["total"] += 1
            except (ValueError, KeyError, TypeError) as e:
                self._stats["invalid"] += 1
                print(f"跳过无效记录: {raw}，原因: {e}")
        return results

    @property
    def stats(self) -> Dict:
        return self._stats

# 最小调用入口（演示 CSV 源）
if __name__ == "__main__":
    import tempfile, os
    
    # 构造演示数据
    demo_csv = os.path.join(tempfile.gettempdir(), "demo_ts.csv")
    with open(demo_csv, "w", newline='') as f:
        writer = csv.writer(f)
        writer.writerow(["timestamp", "value", "device"])
        writer.writerow(["2024-01-01T00:00:00Z", "23.5", "sensor_a"])
        writer.writerow(["2024-01-01T00:01:00Z", "24.1", "sensor_a"])
        writer.writerow(["invalid-time", "25.0", "sensor_a"])  # 无效行
    
    pipeline = IngestionPipeline(CSVSource(demo_csv, tag_cols=["device"]))
    records = pipeline.run()
    print(f"有效记录: {len(records)}，无效记录: {pipeline.stats['invalid']}")
    for r in records[:2]:
        print(r)

运行方式

python ingestion_pipeline.py

验证方法

• 功能验证：运行后应输出 2 条有效记录（时间戳已转为 UTC 格式），并打印 1 条“跳过无效记录”提示（因 invalid-time 不符合 ISO 8601）。
• 扩展性验证：新增数据源时继承 DataSource 并实现 read_raw()，传入 IngestionPipeline 即可，无需修改管道逻辑。
• 完整性验证：若记录缺少 timestamp 或 value 字段，会触发 KeyError 被捕获并计入无效记录，不会中断管道。

预期输出（确定性结果）

跳过无效记录: {'timestamp': 'invalid-time', 'value': '25.0', 'tags': {'device': 'sensor_a'}}，原因: 非法时间戳格式: invalid-time
有效记录: 2，无效记录: 1
{'timestamp': '2024-01-01T00:00:00+00:00', 'value': 23.5, 'tags': {'device': 'sensor_a'}}
{'timestamp': '2024-01-01T00:01:00+00:00', 'value': 24.1, 'tags': {'device': 'sensor_a'}}

API 源接入说明

使用 APISource 时，需确保接口返回 JSON 数组或包含 data 键的对象，且每个元素含 timestamp（ISO 8601 带时区）与 value 字段。例如：

pipeline = IngestionPipeline(APISource("https://api.example.com/metrics", {"limit": 100}))
records = pipeline.run()