# 创建时序数据表
CREATE TABLE DEVICE_DATA (
DATA_ID BIGINT NOT NULL GENERATED ALWAYS AS IDENTITY,
DEVICE_ID VARCHAR(32) NOT NULL,
PARAM_ID VARCHAR(32) NOT NULL,
COLLECT_TIME TIMESTAMP NOT NULL,
PARAM_VALUE DECIMAL(20, 6),
QUALITY INTEGER,
CREATE_TIME TIMESTAMP NOT NULL DEFAULT CURRENT TIMESTAMP,
CONSTRAINT PK_DEVICE_DATA PRIMARY KEY (DATA_ID)
) PARTITION BY RANGE (COLLECT_TIME)
(
STARTING ‘2026-01-01-00.00.00.000000’
ENDING ‘2026-12-31-23.59.59.999999’
EVERY 1 HOURS
);

# 创建索引
CREATE INDEX IDX_DEVICE_DATA_DEVICE ON DEVICE_DATA(DEVICE_ID, COLLECT_TIME DESC) PARTITIONED;
CREATE INDEX IDX_DEVICE_DATA_TIME ON DEVICE_DATA(COLLECT_TIME DESC) PARTITIONED;

# 创建监控指标表
CREATE TABLE METRIC_DATA (
METRIC_ID BIGINT NOT NULL GENERATED ALWAYS AS IDENTITY,
HOST_NAME VARCHAR(100) NOT NULL,
METRIC_NAME VARCHAR(100) NOT NULL,
COLLECT_TIME TIMESTAMP NOT NULL,
METRIC_VALUE DECIMAL(20, 6),
UNIT VARCHAR(20),
CREATE_TIME TIMESTAMP NOT NULL DEFAULT CURRENT TIMESTAMP,
CONSTRAINT PK_METRIC_DATA PRIMARY KEY (METRIC_ID)
) PARTITION BY RANGE (COLLECT_TIME)
(
STARTING ‘2026-01-01-00.00.00.000000’
ENDING ‘2026-12-31-23.59.59.999999’
EVERY 1 HOURS
);

CREATE INDEX IDX_METRIC_DATA_HOST ON METRIC_DATA(HOST_NAME, METRIC_NAME, COLLECT_TIME DESC) PARTITIONED;

# 批量插入时序数据
INSERT INTO DEVICE_DATA (DEVICE_ID, PARAM_ID, COLLECT_TIME, PARAM_VALUE, QUALITY)
VALUES
(‘DEV001’, ‘TEMP’, ‘2026-04-08 10:00:00’, 25.5, 100),
(‘DEV001’, ‘HUMIDITY’, ‘2026-04-08 10:00:00’, 65.2, 100),
(‘DEV001’, ‘PRESSURE’, ‘2026-04-08 10:00:00’, 1013.25, 100),
(‘DEV002’, ‘TEMP’, ‘2026-04-08 10:00:00’, 26.3, 100),
(‘DEV002’, ‘HUMIDITY’, ‘2026-04-08 10:00:00’, 68.1, 100);

# 使用LOAD批量加载
LOAD FROM /data/device_data.csv OF DEL
INSERT INTO DEVICE_DATA
NONRECOVERABLE;

# 创建存储过程：批量插入
CREATE OR REPLACE PROCEDURE SP_INSERT_DEVICE_DATA(
IN p_device_id VARCHAR(32),
IN p_param_id VARCHAR(32),
IN p_collect_time TIMESTAMP,
IN p_param_value DECIMAL(20, 6),
IN p_quality INTEGER
)
LANGUAGE SQL
BEGIN
INSERT INTO DEVICE_DATA (
DEVICE_ID, PARAM_ID, COLLECT_TIME, PARAM_VALUE, QUALITY
) VALUES (
p_device_id, p_param_id, p_collect_time, p_param_value, p_quality
);
END;

# 查询最近一小时数据
SELECT
DEVICE_ID,
PARAM_ID,
COLLECT_TIME,
PARAM_VALUE,
QUALITY
FROM DEVICE_DATA
WHERE COLLECT_TIME >= CURRENT TIMESTAMP – 1 HOUR
ORDER BY COLLECT_TIME DESC;

# 查询指定时间范围
SELECT
DEVICE_ID,
PARAM_ID,
COLLECT_TIME,
PARAM_VALUE
FROM DEVICE_DATA
WHERE COLLECT_TIME BETWEEN ‘2026-04-08 00:00:00’ AND ‘2026-04-08 23:59:59’
ORDER BY COLLECT_TIME;

# 查询指定设备数据
SELECT
COLLECT_TIME,
PARAM_VALUE,
QUALITY
FROM DEVICE_DATA
WHERE DEVICE_ID = ‘DEV001’
AND PARAM_ID = ‘TEMP’
AND COLLECT_TIME >= CURRENT TIMESTAMP – 24 HOURS
ORDER BY COLLECT_TIME;

# 查询最新值
SELECT
DEVICE_ID,
PARAM_ID,
PARAM_VALUE,
COLLECT_TIME
FROM (
SELECT
DEVICE_ID,
PARAM_ID,
PARAM_VALUE,
COLLECT_TIME,
ROW_NUMBER() OVER (PARTITION BY DEVICE_ID, PARAM_ID ORDER BY COLLECT_TIME DESC) AS RN
FROM DEVICE_DATA
WHERE COLLECT_TIME >= CURRENT TIMESTAMP – 1 HOUR
)
WHERE RN = 1;

# 计算变化率
SELECT
DEVICE_ID,
COLLECT_TIME,
PARAM_VALUE,
LAG(PARAM_VALUE, 1) OVER (ORDER BY COLLECT_TIME) AS PREV_VALUE,
PARAM_VALUE – LAG(PARAM_VALUE, 1) OVER (ORDER BY COLLECT_TIME) AS CHANGE,
ROUND((PARAM_VALUE – LAG(PARAM_VALUE, 1) OVER (ORDER BY COLLECT_TIME))
/ LAG(PARAM_VALUE, 1) OVER (ORDER BY COLLECT_TIME) * 100, 2) AS CHANGE_PERCENT
FROM DEVICE_DATA
WHERE DEVICE_ID = ‘DEV001’
AND PARAM_ID = ‘TEMP’
ORDER BY COLLECT_TIME;

# 计算移动平均
SELECT
COLLECT_TIME,
PARAM_VALUE,
AVG(PARAM_VALUE) OVER (
ORDER BY COLLECT_TIME
ROWS BETWEEN 59 PRECEDING AND CURRENT ROW
) AS MOVING_AVG_60MIN
FROM DEVICE_DATA
WHERE DEVICE_ID = ‘DEV001’
AND PARAM_ID = ‘TEMP’
ORDER BY COLLECT_TIME;

# 检测异常值
SELECT
DEVICE_ID,
COLLECT_TIME,
PARAM_VALUE,
CASE
WHEN PARAM_VALUE > AVG(PARAM_VALUE) OVER () + 3 * STDDEV(PARAM_VALUE) OVER () THEN ‘HIGH’
WHEN PARAM_VALUE < AVG(PARAM_VALUE) OVER () - 3 * STDDEV(PARAM_VALUE) OVER () THEN 'LOW' ELSE 'NORMAL' END AS STATUS FROM DEVICE_DATA WHERE DEVICE_ID = 'DEV001' AND PARAM_ID = 'TEMP' AND COLLECT_TIME >= CURRENT TIMESTAMP – 1 HOUR;

# 按小时聚合
SELECT
DEVICE_ID,
PARAM_ID,
DATE(COLLECT_TIME) AS DATA_DATE,
HOUR(COLLECT_TIME) AS DATA_HOUR,
MIN(PARAM_VALUE) AS MIN_VALUE,
MAX(PARAM_VALUE) AS MAX_VALUE,
AVG(PARAM_VALUE) AS AVG_VALUE,
STDDEV(PARAM_VALUE) AS STDDEV_VALUE,
COUNT(*) AS SAMPLE_COUNT
FROM DEVICE_DATA
WHERE COLLECT_TIME >= CURRENT DATE
GROUP BY DEVICE_ID, PARAM_ID, DATE(COLLECT_TIME), HOUR(COLLECT_TIME)
ORDER BY DATA_DATE, DATA_HOUR;

# 按天聚合
SELECT
DEVICE_ID,
PARAM_ID,
DATE(COLLECT_TIME) AS DATA_DATE,
MIN(PARAM_VALUE) AS MIN_VALUE,
MAX(PARAM_VALUE) AS MAX_VALUE,
AVG(PARAM_VALUE) AS AVG_VALUE,
COUNT(*) AS SAMPLE_COUNT
FROM DEVICE_DATA
WHERE COLLECT_TIME >= CURRENT DATE – 7 DAYS
GROUP BY DEVICE_ID, PARAM_ID, DATE(COLLECT_TIME)
ORDER BY DATA_DATE;

# 创建聚合MQT
CREATE TABLE MQT_DEVICE_DATA_HOUR AS (
SELECT
DEVICE_ID,
PARAM_ID,
DATE(COLLECT_TIME) AS DATA_DATE,
HOUR(COLLECT_TIME) AS DATA_HOUR,
MIN(PARAM_VALUE) AS MIN_VALUE,
MAX(PARAM_VALUE) AS MAX_VALUE,
AVG(PARAM_VALUE) AS AVG_VALUE,
COUNT(*) AS SAMPLE_COUNT
FROM DEVICE_DATA
GROUP BY DEVICE_ID, PARAM_ID, DATE(COLLECT_TIME), HOUR(COLLECT_TIME)
) DATA INITIALLY DEFERRED REFRESH DEFERRED
ORGANIZE BY COLUMN;

REFRESH TABLE MQT_DEVICE_DATA_HOUR;

# 每10分钟采样一次
SELECT
DEVICE_ID,
PARAM_ID,
TIMESTAMP(
DATE(COLLECT_TIME),
TIME(’00:00:00′) + (HOUR(COLLECT_TIME) * 60 + MINUTE(COLLECT_TIME)) / 10 * 10 MINUTES
) AS SAMPLE_TIME,
AVG(PARAM_VALUE) AS AVG_VALUE
FROM DEVICE_DATA
WHERE COLLECT_TIME >= CURRENT TIMESTAMP – 1 HOUR
GROUP BY DEVICE_ID, PARAM_ID,
DATE(COLLECT_TIME),
(HOUR(COLLECT_TIME) * 60 + MINUTE(COLLECT_TIME)) / 10
ORDER BY SAMPLE_TIME;

# 创建降采样存储过程
CREATE OR REPLACE PROCEDURE SP_DOWNSAMPLE_DEVICE_DATA(
IN p_start_time TIMESTAMP,
IN p_end_time TIMESTAMP,
IN p_interval_minutes INTEGER
)
LANGUAGE SQL
BEGIN
INSERT INTO DEVICE_DATA_DOWNSAMPLED (
DEVICE_ID, PARAM_ID, SAMPLE_TIME, AVG_VALUE, SAMPLE_COUNT
)
SELECT
DEVICE_ID,
PARAM_ID,
TIMESTAMP(
DATE(COLLECT_TIME),
TIME(’00:00:00′) +
(HOUR(COLLECT_TIME) * 60 + MINUTE(COLLECT_TIME)) / p_interval_minutes * p_interval_minutes MINUTES
) AS SAMPLE_TIME,
AVG(PARAM_VALUE) AS AVG_VALUE,
COUNT(*) AS SAMPLE_COUNT
FROM DEVICE_DATA
WHERE COLLECT_TIME BETWEEN p_start_time AND p_end_time
GROUP BY DEVICE_ID, PARAM_ID,
DATE(COLLECT_TIME),
(HOUR(COLLECT_TIME) * 60 + MINUTE(COLLECT_TIME)) / p_interval_minutes;
END;