4. 数据源集成（一）：关系型数据库 MySQL

在实时监控大屏项目中，MySQL 是最常用的持久化数据源之一。本节将系统讲解从 MySQL 建表、Python 连接（pymysql / SQLAlchemy）到数据读取并转为 DataFrame 的完整链路，为后续数据清洗与可视化提供基础。

4.1 MySQL 建表语句与字段设计

监控大屏的数据通常包含时间戳、指标值、设备/业务标识等核心字段。以下是一个典型的服务器性能监控表建表语句：

CREATE TABLE server_monitor (
    id          BIGINT AUTO_INCREMENT PRIMARY KEY,
    server_ip   VARCHAR(15)     NOT NULL COMMENT '服务器IP',
    cpu_usage   DECIMAL(5,2)    NOT NULL COMMENT 'CPU使用率(%)',
    mem_usage   DECIMAL(5,2)    NOT NULL COMMENT '内存使用率(%)',
    disk_io     DECIMAL(10,2)   NOT NULL COMMENT '磁盘IO (MB/s)',
    net_in      BIGINT          NOT NULL COMMENT '网络入流量 (bytes)',
    net_out     BIGINT          NOT NULL COMMENT '网络出流量 (bytes)',
    status      TINYINT         NOT NULL DEFAULT 1 COMMENT '状态:1正常 0异常',
    record_time DATETIME        NOT NULL COMMENT '采集时间',
    INDEX idx_time (record_time),
    INDEX idx_ip_time (server_ip, record_time)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='服务器实时监控数据表';

4.2 Python 连接 MySQL：两种主流方式

Python 生态中连接 MySQL 最常用的两个库是 pymysql（纯 Python 实现）和 SQLAlchemy（ORM 框架）。下表对比了它们的核心差异：

4.2.1 使用 pymysql 连接

import pymysql

# 建立连接
conn = pymysql.connect(
    host='localhost',
    port=3306,
    user='monitor_user',
    password='your_password',
    database='monitor_db',
    charset='utf8mb4'
)

# 创建游标，返回字典格式
cursor = conn.cursor(pymysql.cursors.DictCursor)

# 执行查询
cursor.execute("SELECT * FROM server_monitor WHERE record_time >= %s", ('2025-01-01 00:00:00',))
rows = cursor.fetchall()

# 关闭连接
cursor.close()
conn.close()

4.2.2 使用 SQLAlchemy 连接

from sqlalchemy import create_engine, text

# 创建引擎（连接池）
engine = create_engine(
    'mysql+pymysql://monitor_user:your_password@localhost:3306/monitor_db?charset=utf8mb4',
    pool_size=5,
    max_overflow=10
)

# 执行原生 SQL
with engine.connect() as conn:
    result = conn.execute(
        text("SELECT * FROM server_monitor WHERE record_time >= :time"),
        {"time": "2025-01-01 00:00:00"}
    )
    rows = result.fetchall()

4.3 读取数据并转为 DataFrame

pandas 的 read_sql() 函数可以直接将 SQL 查询结果转换为 DataFrame，是数据集成中最核心的一步。以下展示两种连接方式下的转换方法：

4.3.1 基于 pymysql 连接

import pandas as pd
import pymysql

conn = pymysql.connect(
    host='localhost', user='monitor_user',
    password='your_password', database='monitor_db',
    charset='utf8mb4'
)

df = pd.read_sql(
    "SELECT * FROM server_monitor WHERE record_time >= %(time)s",
    con=conn,
    params={"time": "2025-01-01 00:00:00"}
)

conn.close()
print(df.head())

4.3.2 基于 SQLAlchemy 引擎（推荐）

import pandas as pd
from sqlalchemy import create_engine, text

engine = create_engine(
    'mysql+pymysql://monitor_user:your_password@localhost:3306/monitor_db?charset=utf8mb4'
)

query = """
    SELECT server_ip, cpu_usage, mem_usage, record_time
    FROM server_monitor
    WHERE record_time BETWEEN :start AND :end
    ORDER BY record_time
"""

df = pd.read_sql(
    sql=text(query),
    con=engine,
    params={"start": "2025-01-01 00:00:00", "end": "2025-01-02 00:00:00"}
)

print(f"读取 {len(df)} 条记录，内存占用约 {df.memory_usage(deep=True).sum() / 1024:.1f} KB")

4.4 数据集成流程架构

下图展示了从 MySQL 到 DataFrame 的完整数据集成链路，包含连接建立、查询执行、结果转换三个核心阶段：

4.5 常见问题与排查