基于Python存储预报时间序列数据的实现方案咨询

不用上专门的时序数据库，也不用每次爬取新建表、给每个指标单独建宽表。你之前方案的核心问题是把动态变化的预报时间戳当成了固定表列，用SQLite最基础的长表结构就能解决所有适配问题，全是你已经掌握的基础语法，没有新概念。

核心表结构设计 #

总共只需要2张表，不需要动态建表、改表：

表1：爬取任务记录表 crawl_tasks #

存每次爬取动作的元信息，替代你之前设计的单独爬取日期表，用自增ID当主键比单纯用日期靠谱，避免一天爬多次、爬取时间跨零点导致的主键冲突：

• crawl_id：整数主键，自增
• crawl_time：时间戳类型，存实际发起爬取的精确时间，加唯一约束防止重复记录同一次爬取
• crawl_status：文本类型，可选，标记本次爬取是否成功，方便后续排查问题

建表SQL：

CREATE TABLE IF NOT EXISTS crawl_tasks (
    crawl_id INTEGER PRIMARY KEY AUTOINCREMENT,
    crawl_time TIMESTAMP NOT NULL UNIQUE,
    crawl_status TEXT DEFAULT 'success'
);

表2：预报数值表 forecast_metrics #

所有指标的预报值全部存在这一张表里，不用为每个指标单独建表，完全适配时间戳不固定、时段不统一的场景：

• record_id：整数主键，自增
• crawl_id：整数，外键关联crawl_tasks表的crawl_id，标记这条预报是哪次爬取拿到的
• forecast_time：时间戳类型，存预报对应的目标时间（比如6月15日爬取拿到的6月16日15点的预报，就存6月16日15点的时间）
• metric_name：文本类型，存指标名称，固定四个值：wind_speed（风速）、gust_wind_speed（阵风风速）、wave_height（浪高）、wave_period（浪周期）
• metric_value：浮点数类型，存对应指标的预报数值

给三个字段(crawl_id, forecast_time, metric_name)加联合唯一约束，从数据库层面杜绝重复数据插入，完全符合关系型数据库避免冗余的设计原则。

建表SQL：

CREATE TABLE IF NOT EXISTS forecast_metrics (
    record_id INTEGER PRIMARY KEY AUTOINCREMENT,
    crawl_id INTEGER NOT NULL,
    forecast_time TIMESTAMP NOT NULL,
    metric_name TEXT NOT NULL,
    metric_value REAL NOT NULL,
    FOREIGN KEY (crawl_id) REFERENCES crawl_tasks(crawl_id),
    UNIQUE(crawl_id, forecast_time, metric_name)
);

方案适配性说明 #

• 完全解决你提到的两个核心问题：不管预报只覆盖凌晨3点到晚上9点的时段，还是受爬取时间影响时间戳间隔不固定，有多少条数据直接插多少行即可，不需要提前预设固定列、不需要修改表结构
• 存储效率远高于按天建表、按指标建宽表的方案：没有空值、没有重复数据，按每天爬1次、每次拿20个预报时间点计算，存10年也就不到3万条记录，SQLite查询是毫秒级响应，完全不会卡顿
• 维护成本极低：所有数据存在单个db文件里，不会出现按天存csv导致的文件散乱、乱码、重复备份的问题

常用查询示例 #

后续做分析时直接写SQL拉数据即可，举两个最常用的场景：

1. 查某一次爬取拿到的所有风速预报

SELECT forecast_time, metric_value
FROM forecast_metrics
WHERE crawl_id = 要查询的爬取ID
  AND metric_name = 'wind_speed'
ORDER BY forecast_time;

2. 对比不同爬取批次对同一个时间点的浪高预报偏差

SELECT t.crawl_time, m.metric_value
FROM forecast_metrics m
JOIN crawl_tasks t ON m.crawl_id = t.crawl_id
WHERE m.forecast_time = '2024-06-20 15:00:00'
  AND m.metric_name = 'wave_height'
ORDER BY t.crawl_time;

极简Python写入实现（用标准库sqlite3，无需额外安装依赖） #

import sqlite3
from datetime import datetime

# 连接数据库，文件不存在会自动创建
conn = sqlite3.connect('weather_forecast.db')
cursor = conn.cursor()

# 首次运行自动建表，后续运行不会重复建
cursor.execute("""
CREATE TABLE IF NOT EXISTS crawl_tasks (
    crawl_id INTEGER PRIMARY KEY AUTOINCREMENT,
    crawl_time TIMESTAMP NOT NULL UNIQUE,
    crawl_status TEXT DEFAULT 'success'
)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS forecast_metrics (
    record_id INTEGER PRIMARY KEY AUTOINCREMENT,
    crawl_id INTEGER NOT NULL,
    forecast_time TIMESTAMP NOT NULL,
    metric_name TEXT NOT NULL,
    metric_value REAL NOT NULL,
    FOREIGN KEY (crawl_id) REFERENCES crawl_tasks(crawl_id),
    UNIQUE(crawl_id, forecast_time, metric_name)
)
""")
conn.commit()

# --------------------------
# 这里写你的爬虫逻辑，最终拿到结构化的预报数据
# 假设爬取结果格式为：
# forecast_data = [
#     {"forecast_time": "2024-06-15 03:00:00", "wind_speed": 3.2, "gust_wind_speed": 5.1, "wave_height": 0.8, "wave_period": 4.2},
#     ... 其余时间点的预报数据
# ]
# --------------------------

current_crawl_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# 写入本次爬取记录，重复则自动跳过
cursor.execute(
    "INSERT OR IGNORE INTO crawl_tasks (crawl_time) VALUES (?)",
    (current_crawl_time,)
)
crawl_id = cursor.lastrowid
# 如果是重复爬取，就查已有的crawl_id
if crawl_id == 0:
    cursor.execute("SELECT crawl_id FROM crawl_tasks WHERE crawl_time = ?", (current_crawl_time,))
    crawl_id = cursor.fetchone()[0]

# 批量拼装所有指标的插入数据
insert_rows = []
for point in forecast_data:
    ft = point["forecast_time"]
    insert_rows.append((crawl_id, ft, "wind_speed", point["wind_speed"]))
    insert_rows.append((crawl_id, ft, "gust_wind_speed", point["gust_wind_speed"]))
    insert_rows.append((crawl_id, ft, "wave_height", point["wave_height"]))
    insert_rows.append((crawl_id, ft, "wave_period", point["wave_period"]))

# 批量写入，重复数据自动跳过，不会报错
cursor.executemany(
    "INSERT OR IGNORE INTO forecast_metrics (crawl_id, forecast_time, metric_name, metric_value) VALUES (?, ?, ?, ?)",
    insert_rows
)
conn.commit()
conn.close()

后续用pandas做分析时，直接调用pd.read_sql()写查询语句就能把结果转成DataFrame，和你平时处理csv的流程完全兼容。不用盲目跟风上专用时序数据库，对你这个数据量级来说，SQLite是维护成本最低、最不容易出问题的选择。

内容的提问来源于stack exchange，提问作者rincon