SQL时间序列数据按指定间隔采样及插值实现咨询

这个需求既可以用纯SQL实现，也可以通过Python做后处理，两种方案各有适用场景，具体如下：

核心步骤 #

1. 生成指定间隔的时间基准序列
   先确定数据集的时间范围（最小/最大测量时间），用递归CTE生成每隔8秒的时间点，作为采样的基准框架。以PostgreSQL为例：

   WITH time_series AS (
       SELECT MIN(measure_time) AS ts FROM device_data
       UNION ALL
       SELECT ts + INTERVAL '8 seconds' FROM time_series
       WHERE ts + INTERVAL '8 seconds' <= (SELECT MAX(measure_time) FROM device_data)
   )
   SELECT ts FROM time_series;
   

   不同数据库语法略有差异：MySQL需加WITH RECURSIVE，SQL Server用DATEADD(second, 8, ts)，Oracle用ts + NUMTODSINTERVAL(8, 'SECOND')。

2. 关联设备数据并完成插值
   将基准时间序列与设备数据按设备ID关联，用窗口函数LAG()/LEAD()获取前后最近的有效测量值，再根据时间差计算插值（以下示例为线性插值）：

   WITH time_series AS (
       -- 上述基准时间生成逻辑
       SELECT MIN(measure_time) AS ts FROM device_data
       UNION ALL
       SELECT ts + INTERVAL '8 seconds' FROM time_series
       WHERE ts + INTERVAL '8 seconds' <= (SELECT MAX(measure_time) FROM device_data)
   ),
   device_base AS (
       SELECT DISTINCT device_id FROM device_data
   ),
   joined_data AS (
       SELECT 
           db.device_id,
           ts.ts,
           dd.measure_value,
           LAG(dd.measure_value) OVER (PARTITION BY db.device_id ORDER BY ts.ts) AS prev_val,
           LAG(dd.measure_time) OVER (PARTITION BY db.device_id ORDER BY ts.ts) AS prev_time,
           LEAD(dd.measure_value) OVER (PARTITION BY db.device_id ORDER BY ts.ts) AS next_val,
           LEAD(dd.measure_time) OVER (PARTITION BY db.device_id ORDER BY ts.ts) AS next_time
       FROM time_series ts
       CROSS JOIN device_base db
       LEFT JOIN device_data dd 
           ON dd.device_id = db.device_id 
           AND dd.measure_time BETWEEN ts.ts - INTERVAL '8 seconds' AND ts.ts
   )
   SELECT 
       device_id,
       ts,
       CASE
           WHEN measure_value IS NOT NULL THEN measure_value
           ELSE prev_val + (next_val - prev_val) * EXTRACT(EPOCH FROM (ts - prev_time)) / EXTRACT(EPOCH FROM (next_time - prev_time))
       END AS interpolated_value
   FROM joined_data;
   

   你可以根据需求替换插值逻辑，比如用prev_val直接填充最近值，无需计算线性插值。

核心流程 #

用pandas处理时间序列更直观，适合复杂插值规则或后续需做分析的场景：

1. 从数据库读取原始数据，将时间列转为datetime类型并按设备分组
2. 按8秒重采样，调用插值方法生成结果
   示例代码：

import pandas as pd
import sqlalchemy

# 连接数据库并读取数据
engine = sqlalchemy.create_engine('your_db_connection_string')
df = pd.read_sql('SELECT device_id, measure_time, measure_value FROM device_data', engine)

# 预处理：转换时间格式，设置索引
df['measure_time'] = pd.to_datetime(df['measure_time'])
df = df.set_index('measure_time').groupby('device_id').resample('8S').asfreq()

# 线性插值，可替换为'nearest'（最近值填充）等方法
df['interpolated_value'] = df['measure_value'].interpolate(method='linear')
df = df.reset_index()

• 若数据量较大、希望在数据库层完成处理（减少数据传输开销），优先用SQL实现
• 若插值逻辑复杂（如非线性插值、自定义规则），或后续需做可视化/深度分析，Python方案更灵活

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