如何优化解析JSON数据的MySQL查询及聚合表性能

问题背景 #

我有一张Analytics表，结构如下：

+---------+----------+----------------+------------+---------------------+-------+
| user_id | label_id | application_id | session_id | date                | score |
+---------+----------+----------------+------------+---------------------+-------+
|       1 |      123 |              1 | ABCD123456 | 2023-06-15 09:05:18 |   123 |
|       1 |      456 |              1 | ABCD123456 | 2023-06-15 09:05:18 |   456 |
|       1 |      789 |              1 | ABCD123456 | 2023-06-15 09:05:18 |   789 |
+---------+----------+----------------+------------+---------------------+-------+

score列类型为varchar，开发中途改为存储JSON数据。为提升报表生成速度，我创建了类物化视图的表WeldingSessionsAgregated（通过触发器维护），报表速度提升10倍以上，但批量插入会话时该表操作极耗时，仅处理150条会话就导致CPU占用100%，需优化代码适配高并发场景。

优化方案 #

1. 添加关键索引，消除全表扫描 #

当前Analytics表缺少查询高频的联合索引，大量按session_id+label_id的查询会走全表扫描，直接拖慢性能。

-- 给Analytics添加会话+标签的联合索引
ALTER TABLE Analytics ADD INDEX idx_session_label (session_id, label_id);
-- 给聚合表的会话ID加主键，加速更新/删除操作
ALTER TABLE WeldingSessionsAgregated ADD PRIMARY KEY (s_id);

2. 重写求和函数，避免循环查询 #

原get_welding_array_sum函数每次循环都查询一次数据库，性能极低。改用MySQL原生JSON_TABLE函数一次性解析数组并求和：

DROP FUNCTION IF EXISTS get_welding_array_sum;
CREATE FUNCTION get_welding_array_sum(sid VARCHAR(256), array_name VARCHAR(256))
RETURNS DECIMAL(10,2)
BEGIN
    DECLARE result DECIMAL(10,2);
    SELECT COALESCE(SUM(val), 0.0) INTO result
    FROM Analytics a,
         JSON_TABLE(
             a.score,
             CONCAT('$.', array_name, '[*]') COLUMNS(val DECIMAL(10,2) PATH '$')
         ) jt
    WHERE a.label_id = 127 AND a.session_id = sid;
    RETURN result;
END;

3. 简化触发器逻辑，减少重复查询 #

原触发器拆分多个UPDATE语句且重复查询相同数据，合并逻辑并减少查询次数：

DROP TRIGGER IF EXISTS WeldingSessionsAgregatedInsertTrigger;
CREATE TRIGGER WeldingSessionsAgregatedInsertTrigger AFTER INSERT ON Analytics
FOR EACH ROW
BEGIN
    DECLARE existing_count INT;
    DECLARE org_id INT;
    DECLARE is_sealing INT;
    
    IF NEW.application_id = 7 THEN
        -- 一次性查询是否为sealing会话
        SELECT COUNT(*) INTO is_sealing FROM Analytics WHERE label_id = 134 AND session_id = NEW.session_id AND score = "sealing";
        
        IF is_sealing > 0 THEN
            DELETE FROM WeldingSessionsAgregated WHERE s_id = NEW.session_id;
        ELSE
            -- 查询聚合表中是否已存在该会话
            SELECT COUNT(*) INTO existing_count FROM WeldingSessionsAgregated WHERE s_id = NEW.session_id;
            
            IF existing_count = 0 THEN
                -- 一次性获取组织ID
                SELECT organization_id INTO org_id FROM Users WHERE id = NEW.user_id;
                INSERT INTO WeldingSessionsAgregated (
                    s_id, user_id, date, application_id, organization_id
                ) VALUES (
                    NEW.session_id, NEW.user_id, NEW.date, NEW.application_id, org_id
                );
            END IF;
            
            -- 用CASE合并UPDATE操作，避免多次触发更新
            CASE NEW.label_id
                WHEN 125 THEN UPDATE WeldingSessionsAgregated SET session_start = NEW.score WHERE s_id = NEW.session_id;
                WHEN 126 THEN UPDATE WeldingSessionsAgregated SET session_end = NEW.score WHERE s_id = NEW.session_id;
                WHEN 129 THEN UPDATE WeldingSessionsAgregated SET part_name = NEW.score WHERE s_id = NEW.session_id;
                WHEN 132 THEN UPDATE WeldingSessionsAgregated SET displacement_map = NEW.score WHERE s_id = NEW.session_id;
                WHEN 133 THEN UPDATE WeldingSessionsAgregated SET weld_preview = NEW.score WHERE s_id = NEW.session_id;
                WHEN 135 THEN UPDATE WeldingSessionsAgregated SET method = NEW.score WHERE s_id = NEW.session_id;
                WHEN 136 THEN UPDATE WeldingSessionsAgregated SET material_used = NEW.score WHERE s_id = NEW.session_id;
                WHEN 127 THEN
                    -- 一次性计算所有聚合值，减少函数调用次数
                    UPDATE WeldingSessionsAgregated
                    SET
                        speed_sum = get_welding_array_sum(NEW.session_id, "speedEntries"),
                        travel_angle_sum = get_welding_array_sum(NEW.session_id, "travelAngleEntries"),
                        work_angle_sum = get_welding_array_sum(NEW.session_id, "workAngleEntries"),
                        distance_sum = get_welding_array_sum(NEW.session_id, "distanceEntries"),
                        count = JSON_LENGTH(NEW.score, '$.speedEntries'),
                        avg_speed = get_welding_array_sum(NEW.session_id, "speedEntries") / (JSON_LENGTH(NEW.score, '$.speedEntries') + 0.00001),
                        avg_travel_angle = get_welding_array_sum(NEW.session_id, "travelAngleEntries") / (JSON_LENGTH(NEW.score, '$.travelAngleEntries') + 0.00001),
                        avg_work_angle = get_welding_array_sum(NEW.session_id, "workAngleEntries") / (JSON_LENGTH(NEW.score, '$.workAngleEntries') + 0.00001),
                        avg_distance = get_welding_array_sum(NEW.session_id, "distanceEntries") / (JSON_LENGTH(NEW.score, '$.distanceEntries') + 0.00001)
                    WHERE s_id = NEW.session_id;
            END CASE;
        END IF;
    END IF;
END; //

4. 重建物化视图时使用批量聚合查询 #

原重建表语句使用大量子查询，效率极低。改用GROUP BY聚合查询一次性生成数据：

DROP TABLE IF EXISTS WeldingSessionsAgregated;
CREATE TABLE WeldingSessionsAgregated AS
SELECT
    a.session_id AS s_id,
    a.user_id,
    a.date,
    a.application_id,
    MAX(CASE WHEN a.label_id = 125 THEN a.score END) AS session_start,
    MAX(CASE WHEN a.label_id = 126 THEN a.score END) AS session_end,
    MAX(CASE WHEN a.label_id = 129 THEN a.score END) AS part_name,
    MAX(CASE WHEN a.label_id = 132 THEN a.score END) AS displacement_map,
    MAX(CASE WHEN a.label_id = 133 THEN a.score END) AS weld_preview,
    MAX(CASE WHEN a.label_id = 135 THEN a.score END) AS method,
    MAX(CASE WHEN a.label_id = 136 THEN a.score END) AS material_used,
    u.organization_id,
    -- 直接计算求和值，避免调用函数
    COALESCE(
        (SELECT SUM(val) FROM JSON_TABLE(aa.score, '$.speedEntries[*]' COLUMNS(val DECIMAL(10,2) PATH '$')) jt),
        0.0
    ) AS speed_sum,
    COALESCE(
        (SELECT SUM(val) FROM JSON_TABLE(aa.score, '$.travelAngleEntries[*]' COLUMNS(val DECIMAL(10,2) PATH '$')) jt),
        0.0
    ) AS travel_angle_sum,
    COALESCE(
        (SELECT SUM(val) FROM JSON_TABLE(aa.score, '$.workAngleEntries[*]' COLUMNS(val DECIMAL(10,2) PATH '$')) jt),
        0.0
    ) AS work_angle_sum,
    COALESCE(
        (SELECT SUM(val) FROM JSON_TABLE(aa.score, '$.distanceEntries[*]' COLUMNS(val DECIMAL(10,2) PATH '$')) jt),
        0.0
    ) AS distance_sum,
    JSON_LENGTH(aa.score, '$.speedEntries') AS count,
    -- 计算平均值
    COALESCE(
        (SELECT SUM(val) FROM JSON_TABLE(aa.score, '$.speedEntries[*]' COLUMNS(val DECIMAL(10,2) PATH '$')) jt) / (JSON_LENGTH(aa.score, '$.speedEntries') + 0.00001),
        0.0
    ) AS avg_speed,
    COALESCE(
        (SELECT SUM(val) FROM JSON_TABLE(aa.score, '$.travelAngleEntries[*]' COLUMNS(val DECIMAL(10,2) PATH '$')) jt) / (JSON_LENGTH(aa.score, '$.travelAngleEntries') + 0.00001),
        0.0
    ) AS avg_travel_angle,
    COALESCE(
        (SELECT SUM(val) FROM JSON_TABLE(aa.score, '$.workAngleEntries[*]' COLUMNS(val DECIMAL(10,2) PATH '$')) jt) / (JSON_LENGTH(aa.score, '$.workAngleEntries') + 0.00001),
        0.0
    ) AS avg_work_angle,
    COALESCE(
        (SELECT SUM(val) FROM JSON_TABLE(aa.score, '$.distanceEntries[*]' COLUMNS(val DECIMAL(10,2) PATH '$')) jt) / (JSON_LENGTH(aa.score, '$.distanceEntries') + 0.00001),
        0.0
    ) AS avg_distance
FROM Analytics a
LEFT JOIN Analytics aa ON a.session_id = aa.session_id AND aa.label_id = 127
LEFT JOIN Users u ON a.user_id = u.id
WHERE a.application_id = 7
GROUP BY a.session_id, a.user_id, a.date, a.application_id, u.organization_id, aa.score;

-- 添加主键索引
ALTER TABLE WeldingSessionsAgregated ADD PRIMARY KEY (s_id);

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