Python 实时数据处理平台架构与实现

Python 实时数据处理平台架构与实现 | 极客日志

import asyncio
from kafka import KafkaProducer
import json
from typing import Dict, Any

class DataCollector:
    def __init__(self, kafka_servers: list):
        self.producer = KafkaProducer(
            bootstrap_servers=kafka_servers,
            value_serializer=lambda v: json.dumps(v).encode('utf-8'),
            compression_type='gzip',
            batch_size=16384,
            linger_ms=10
        )

    async def collect_from_api(self, api_url: str, topic: str):
        """从 API 接口采集数据"""
        async with aiohttp.ClientSession() as session:
            while True:
                try:
                    async with session.get(api_url) as response:
                        data = await response.json()
                        self.send_to_kafka(topic, data)
                        await asyncio.sleep(1)
                except Exception as e:
                    print(f"采集错误：{e}")
                    await asyncio.sleep(5)

    def send_to_kafka(self, topic: str, data: Dict[Any, Any]):
        """发送数据到 Kafka"""
        try:
            self.producer.send(topic, value=data)
            self.producer.flush()
        except Exception as e:
            print(f"发送失败：{e}")

from kafka import KafkaConsumer, KafkaProducer
from datetime import datetime
import json

class StreamProcessor:
    def __init__(self, input_topic: str, output_topic: str):
        self.consumer = KafkaConsumer(
            input_topic,
            bootstrap_servers=['localhost:9092'],
            value_deserializer=lambda m: json.loads(m.decode('utf-8')),
            auto_offset_reset='latest',
            enable_auto_commit=True
        )
        self.producer = KafkaProducer(
            bootstrap_servers=['localhost:9092'],
            value_serializer=lambda v: json.dumps(v).encode('utf-8')
        )
        self.output_topic = output_topic

    def process_data(self, data: dict) -> dict:
        """数据处理逻辑"""
        cleaned_data = self.clean_data(data)
        transformed_data = self.transform_data(cleaned_data)
        aggregated_data = self.aggregate_data(transformed_data)
        aggregated_data['processed_at'] = datetime.now().isoformat()
        return aggregated_data

    def clean_data(self, data: dict) -> dict:
        """数据清洗：去除空值、异常值"""
        return {k: v for k, v in data.items() if v is not None}

    def transform_data(self, data: dict) -> dict:
        """数据转换：格式标准化"""
        if 'temperature' in data:
            data['temperature_celsius'] = (data['temperature'] - 32) * 5/9
        return data

    def aggregate_data(self, data: dict) -> dict:
        """数据聚合：计算统计指标"""
        return data

    def run(self):
        """启动流处理"""
        print("流处理引擎启动...")
        for message in self.consumer:
            try:
                processed_data = self.process_data(message.value)
                self.producer.send(self.output_topic, processed_data)
            except Exception as e:
                print(f"处理错误：{e}")

from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS
from datetime import datetime

class TimeSeriesStorage:
    def __init__(self, url: str, token: str, org: str, bucket: str):
        self.client = InfluxDBClient(url=url, token=token, org=org)
        self.write_api = self.client.write_api(write_options=SYNCHRONOUS)
        self.query_api = self.client.query_api()
        self.bucket = bucket
        self.org = org

    def write_data(self, measurement: str, tags: dict, fields: dict):
        """写入时序数据"""
        point = Point(measurement)
        for tag_key, tag_value in tags.items():
            point.tag(tag_key, tag_value)
        for field_key, field_value in fields.items():
            point.field(field_key, field_value)
        point.time(datetime.utcnow())
        self.write_api.write(bucket=self.bucket, record=point)

    def query_data(self, measurement: str, time_range: str = '-1h'):
        """查询时序数据"""
        query = f''' from(bucket: "{self.bucket}") |> range(start: {time_range}) |> filter(fn: (r) => r._measurement == "{measurement}") '''
        tables = self.query_api.query(query, org=self.org)
        results = []
        for table in tables:
            for record in table.records:
                results.append({
                    'time': record.get_time(),
                    'measurement': record.get_measurement(),
                    'field': record.get_field(),
                    'value': record.get_value(),
                    'tags': record.values
                })
        return results

    def close(self):
        """关闭连接"""
        self.client.close()

from fastapi import FastAPI, WebSocket, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import List, Optional
import asyncio
import json

app = FastAPI(title="实时数据处理平台 API")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

class DataPoint(BaseModel):
    timestamp: str
    metric: str
    value: float
    tags: Optional[dict] = {}

class QueryRequest(BaseModel):
    measurement: str
    time_range: str = '-1h'
    filters: Optional[dict] = {}

@app.get("/api/metrics/latest")
async def get_latest_metrics():
    """获取最新指标数据"""
    return {
        "cpu_usage": 75.5,
        "memory_usage": 68.2,
        "disk_io": 1024,
        "network_traffic": 2048
    }

@app.post("/api/query")
async def query_timeseries(request: QueryRequest):
    """查询时序数据"""
    storage = TimeSeriesStorage(
        url="http://localhost:8086",
        token="your-token",
        org="your-org",
        bucket="your-bucket"
    )
    try:
        results = storage.query_data(
            measurement=request.measurement,
            time_range=request.time_range
        )
        return {"data": results}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))
    finally:
        storage.close()

@app.websocket("/ws/realtime")
async def websocket_endpoint(websocket: WebSocket):
    """WebSocket 实时数据推送"""
    await websocket.accept()
    try:
        while True:
            data = {
                "timestamp": datetime.now().isoformat(),
                "metrics": {
                    "cpu": 75.5,
                    "memory": 68.2,
                    "requests_per_second": 1500
                }
            }
            await websocket.send_json(data)
            await asyncio.sleep(1)
    except Exception as e:
        print(f"WebSocket 错误：{e}")
    finally:
        await websocket.close()

@app.get("/api/statistics/summary")
async def get_statistics():
    """获取统计摘要"""
    return {
        "total_events": 1500000,
        "events_per_second": 1500,
        "active_sources": 25,
        "processing_latency_ms": 45
    }

<template>
  <div class="realtime-dashboard">
    <div class="header">
      <h1>实时数据监控平台</h1>
      <div class="stats">
        <div class="stat-item">
          <span class="label">实时事件数</span>
          <span class="value">{{ stats.eventsPerSecond }}/s</span>
        </div>
        <div class="stat-item">
          <span class="label">活跃数据源</span>
          <span class="value">{{ stats.activeSources }}</span>
        </div>
        <div class="stat-item">
          <span class="label">处理延迟</span>
          <span class="value">{{ stats.latency }}ms</span>
        </div>
      </div>
    </div>
    <div class="charts-container">
      <div class="chart-box"><div ref="cpuChart"></div></div>
      <div class="chart-box"><div ref="memoryChart"></div></div>
      <div class="chart-box"><div ref="trafficChart"></div></div>
    </div>
  </div>
</template>

<script setup>
import { ref, onMounted, onUnmounted } from 'vue'
import * as echarts from 'echarts'

const cpuChart = ref(null)
const memoryChart = ref(null)
const trafficChart = ref(null)
const stats = ref({ eventsPerSecond: 0, activeSources: 0, latency: 0 })
let ws = null
let charts = {}

const initCharts = () => {
  charts.cpu = echarts.init(cpuChart.value)
  charts.cpu.setOption({
    title: { text: 'CPU 使用率', left: 'center' },
    tooltip: { trigger: 'axis' },
    xAxis: { type: 'time', splitLine: { show: false } },
    yAxis: { type: 'value', max: 100, axisLabel: { formatter: '{value}%' } },
    series: [{ name: 'CPU', type: 'line', smooth: true, data: [], areaStyle: { opacity: 0.3 } }]
  })
  // 内存和网络图表初始化类似
}

const connectWebSocket = () => {
  ws = new WebSocket('ws://localhost:8000/ws/realtime')
  ws.onmessage = (event) => {
    const data = JSON.parse(event.data)
    updateCharts(data)
    updateStats(data)
  }
}

const updateCharts = (data) => {
  const timestamp = new Date(data.timestamp)
  const maxDataPoints = 50
  const cpuOption = charts.cpu.getOption()
  cpuOption.series[0].data.push([timestamp, data.metrics.cpu])
  if (cpuOption.series[0].data.length > maxDataPoints) {
    cpuOption.series[0].data.shift()
  }
  charts.cpu.setOption(cpuOption)
}

onMounted(() => {
  initCharts()
  connectWebSocket()
})
onUnmounted(() => {
  if (ws) ws.close()
  Object.values(charts).forEach(chart => chart.dispose())
})
</script>

<style scoped>
.realtime-dashboard { padding: 20px; background: #0a0e27; color: #fff; min-height: 100vh; }
.header h1 { text-align: center; font-size: 32px; margin-bottom: 20px; }
.stats { display: flex; justify-content: center; gap: 40px; }
.stat-item .value { font-size: 24px; font-weight: bold; color: #00d4ff; }
.charts-container { display: grid; grid-template-columns: repeat(auto-fit, minmax(400px, 1fr)); gap: 20px; }
.chart-box { background: #151932; border-radius: 8px; padding: 20px; }
</style>

from dataclasses import dataclass
from enum import Enum

class AlertLevel(Enum):
    INFO = "info"
    WARNING = "warning"
    ERROR = "error"
    CRITICAL = "critical"

@dataclass
class Alert:
    level: AlertLevel
    message: str
    metric: str
    value: float
    threshold: float

class AlertManager:
    def __init__(self):
        self.thresholds = {
            'cpu_usage': 80.0,
            'memory_usage': 85.0,
            'processing_latency': 1000.0,
            'error_rate': 0.05
        }

    def check_metrics(self, metrics: dict):
        alerts = []
        for metric, value in metrics.items():
            if metric in self.thresholds:
                threshold = self.thresholds[metric]
                if value > threshold:
                    alert = Alert(level=self._determine_alert_level(value, threshold), message=f"{metric}超过阈值", metric=metric, value=value, threshold=threshold)
                    alerts.append(alert)
        return alerts

    def _determine_alert_level(self, value: float, threshold: float) -> AlertLevel:
        ratio = value / threshold
        if ratio > 1.5:
            return AlertLevel.CRITICAL
        elif ratio > 1.2:
            return AlertLevel.ERROR
        else:
            return AlertLevel.WARNING

import logging
import json
from datetime import datetime

class StructuredLogger:
    def __init__(self, name: str):
        self.logger = logging.getLogger(name)
        self.logger.setLevel(logging.INFO)
        handler = logging.StreamHandler()
        handler.setFormatter(self.JsonFormatter())
        self.logger.addHandler(handler)

    class JsonFormatter(logging.Formatter):
        def format(self, record):
            log_data = {
                'timestamp': datetime.utcnow().isoformat(),
                'level': record.levelname,
                'logger': record.name,
                'message': record.getMessage(),
                'module': record.module,
                'function': record.funcName,
                'line': record.lineno
            }
            if hasattr(record, 'extra_data'):
                log_data.update(record.extra_data)
            return json.dumps(log_data)

    def info(self, message: str, **kwargs):
        self.logger.info(message, extra={'extra_data': kwargs})

    def error(self, message: str, **kwargs):
        self.logger.error(message, extra={'extra_data': kwargs})

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

version: '3.8'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:latest
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
  kafka:
    image: confluentinc/cp-kafka:latest
    depends_on: [zookeeper]
    ports: ["9092:9092"]
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
  redis:
    image: redis:alpine
    ports: ["6379:6379"]
  influxdb:
    image: influxdb:2.7
    ports: ["8086:8086"]
    environment:
      DOCKER_INFLUXDB_INIT_MODE: setup
      DOCKER_INFLUXDB_INIT_USERNAME: admin
      DOCKER_INFLUXDB_INIT_PASSWORD: adminpassword
  api:
    build: ./backend
    ports: ["8000:8000"]
    depends_on: [kafka, redis, influxdb]
  frontend:
    build: ./frontend
    ports: ["3000:80"]
    depends_on: [api]

apiVersion: apps/v1
kind: Deployment
metadata:
  name: data-platform-api
spec:
  replicas: 3
  selector:
    matchLabels:
      app: data-platform-api
  template:
    metadata:
      labels:
        app: data-platform-api
    spec:
      containers:
      - name: api
        image: data-platform-api:latest
        ports:
        - containerPort: 8000
        resources:
          requests:
            memory: "512Mi"
            cpu: "500m"
          limits:
            memory: "1Gi"
            cpu: "1000m"
---
apiVersion: v1
kind: Service
metadata:
  name: data-platform-api-service
spec:
  selector:
    app: data-platform-api
  ports:
  - protocol: TCP
    port: 80
    targetPort: 8000
  type: LoadBalancer

Python 实时数据处理平台架构与实现

项目概述

技术架构

整体架构设计

更多推荐文章

相关免费在线工具

核心技术栈

核心功能实现

1. 数据采集模块

2. 实时数据处理

3. 数据存储服务

4. FastAPI 服务层

5. 前端实时可视化

性能优化策略

1. 数据处理优化

2. 存储优化

3. 查询优化

监控与运维

1. 系统监控指标

2. 告警机制

3. 日志管理

部署方案

1. 容器化部署

2. Kubernetes 部署

扩展性考虑

1. 水平扩展

2. 垂直扩展

总结与展望

Python 实时数据处理平台架构与实现

项目概述

技术架构

整体架构设计

微信扫一扫，关注极客日志

更多推荐文章

相关免费在线工具

核心技术栈

核心功能实现

1. 数据采集模块

2. 实时数据处理

3. 数据存储服务

4. FastAPI 服务层

5. 前端实时可视化

性能优化策略

1. 数据处理优化

2. 存储优化

3. 查询优化

监控与运维

1. 系统监控指标

2. 告警机制

3. 日志管理

部署方案

1. 容器化部署

2. Kubernetes 部署

扩展性考虑

1. 水平扩展

2. 垂直扩展

总结与展望