X-DOC：Mysql数据库大量数据查询加速（定时JOB和存储过程应用案例）
- 1、案例背景
- 2、解决思路
- 3、实现方式
- 3.1 开启定时调度功能
- 3.2 创建JOB日志表
- 3.3 创建JOB任务
- 3.4 创建JOB
- 3.5 JOB的维护及查看
- 4、总结
1、案例背景
在某中台系统中，设计了大量的基础数据（维度数据、维度映射关系等）来支撑业务功能，业务表中存在大量的维度外键关联字段，其优点是可以实现前端的选择录入，数据校验，确保录入数据的准确性；缺点是在做业务报表时，需要做大量的维度关联（join）操作。
受限于该平台输出报表需要写视图来实现的这一窘境，对于某些较为复杂的报表需求，可能会涉及非常多的表关联，非常影响报表性能，有的甚至需要视图套视图，由于视图数据的全量性，不会根据外部筛选条件优先过滤，更加影响执行效率，对于业务数据量大的甚至会出不来数据。
2、解决思路
引入数据仓库的解决思路，通过定时调度，提前清洗、物化数据，即将某些相对稳定，不常变化的数据，如基础数据、历史业务数据等，通过定时任务，加以必要逻辑处理后，形成中间数据、甚至是最终结果数据，保存到自建的物理表，再在表上创建合理索引，最后基于这些物理表来创建报表，提供给前端用户查询，最终能达到较好的用户体验。
3、实现方式
由于该系统是基于Mysql数据，以下以Mysql为基础，演示实现过程。
3.1 开启定时调度功能
一般关系型数据库，都有自带的定时调度功能，都可以实现定时执行脚本的功能。如SQLServer的作业，Mysql的event。
# 1、开启MYSQL定时设置
-- 1.1、通过show EVENTS显示当前定义的事件
show EVENTS;
（说明：初始列表为空，图中记录为以下步骤操作完成后产生的）
-- 1.2、检查event_scheduler状态
SHOW VARIABLES LIKE 'event_scheduler';
-- 1.3、设置job自动启动可以执行：
SET GLOBAL event_scheduler = ON;
-- 或修改my.ini文件，添加：event_scheduler=1
3.2 创建JOB日志表
自定义一个张表，用来记录Job执行情况，方便后续跟踪。
delimiter #
drop procedure if exists leodb.p_job_log;
create procedure leodb.p_job_log
(
in p_id int,
in p_job varchar(50),
in p_task varchar(50),
in p_note varchar(255)
)
begin
CREATE TABLE if not exists leodb.t_job_log (
id int, -- job id
job varchar(50), -- job名称
task varchar(50), -- 任务名称
starttime datetime, -- 开始时间
endtime datetime, -- 结束时间
NOTE varchar(255), -- 备注信息
primary key( id, job, task)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
-- 存在则更新结束时间，并拼接备注信息
if exists(select 1 from leodb.t_job_log
where id = p_id and job = p_job and task = p_task)
THEN
update leodb.t_job_log set
endtime = now(),
note = case when ifnull(note,'')<>'' then note + '/' + p_note else p_note end
where id = p_id and job = p_job and task = p_task;
-- 不存在则插入日志
else
insert into leodb.t_job_log(id, job, task, starttime, note)
value( p_id, p_job, p_task, now(), p_note );
end if;
end#
delimiter ;
-- 测试日志维护存储过程
-- call leodb.p_job_log( cast(unix_timestamp() as signed) , 'test', 'test', 'test');
-- call leodb.p_job_log( 1688086594,, 'test', 'test', 'test');
3.3 创建JOB任务
# 3、创建具体JOB存储过程
# 3.1、TASK1：维度、维度映射转存物理表
delimiter #
drop procedure if exists leodb.p_job_get_data_4_rp;
create procedure leodb.p_job_get_data_4_rp()
begin
-- ·、成本中心映射预算部门
start transaction;
create table if not exists leodb.t_view_LEOPU_COST_BGDP
(
s_object_code varchar(20),
t_object_code varchar(20)
);
delete from leodb.t_view_LEOPU_COST_BGDP;
insert into leodb.t_view_LEOPU_COST_BGDP
select s_object_code, t_object_code from leodb.view_LEOPU_COST_BGDP;
commit;
-- 2、消费类型映射预算科目
start transaction;
create table if not exists leodb.t_view_LEOPU_EXP_CSM_BGT
(
s_object_code varchar(20),
t_object_code varchar(20)
);
delete from leodb.t_view_LEOPU_EXP_CSM_BGT;
insert into leodb.t_view_LEOPU_EXP_CSM_BGT
select s_object_code, t_object_code from leodb.view_LEOPU_EXP_CSM_BGT;
commit;
-- 3、会计科目映射预算科目
start transaction;
create table if not exists leodb.t_view_LEOPU_ACC_BGT
(
s_object_code varchar(20),
t_object_code varchar(20)
);
delete from leodb.t_view_LEOPU_ACC_BGT;
insert into leodb.t_view_LEOPU_ACC_BGT
select s_object_code, t_object_code from leodb.view_LEOPU_ACC_BGT;
commit;
-- 4、LEOUP付款类型+供应商类型映射贷方（应付报账）
start transaction;
create table if not exists leodb.t_view_LEOPU_PUR_BILL
(
s1_object_code varchar(20),
s2_object_code varchar(20),
t_object_code varchar(20)
);
delete from leodb.t_view_LEOPU_PUR_BILL;
insert into leodb.t_view_LEOPU_PUR_BILL
select s1_object_code, s2_object_code, t_object_code from leodb.view_LEOPU_PUR_BILL;
commit;
end#
delimiter ;
# 3.2、TASK2：提取流程初审人到物理表
-- 基础表：记录流程初审人
select * from leodb.t_mdfp_bpm_audit_first;
drop PROCEDURE if exists leodb.p_job_task_first_audit;
create PROCEDURE leodb.p_job_task_first_audit( in p_minute int )
begin
start transaction;
-- 创建初审人员表
create table if not exists leodb.t_mdfp_bpm_audit_first
( BUSINESS_ID varchar(32), OPERATE_TIME bigint, USER_NAME varchar(255));
-- 创建临时表，减少对审批步骤表的访问，提升查询效率
CREATE TEMPORARY TABLE if not exists leodb.tmp_mdfp_bpm_audit_history(
BUSINESS_ID varchar(32), OPERATE_TIME bigint, USER_NAME varchar(255));
truncate table tmp_mdfp_bpm_audit_history;
-- 拉取最近的初审记录（增量不同：前p_minute分钟内产生的记录）
insert into leodb.tmp_mdfp_bpm_audit_history( BUSINESS_ID, OPERATE_TIME, USER_NAME )
select BUSINESS_ID, OPERATE_TIME, USER_NAME
from mdfp.mdfp_bpm_audit_history as t
where ACT_NAME LIKE '%初审%'
and OPERATE_TYPE = 'approve'
and OPERATE_TIME > UNIX_TIMESTAMP(DATE_ADD(now(),INTERVAL -p_minute MINUTE))*1000
and not exists(select 1 from mdfp.mdfp_bpm_audit_history
where BUSINESS_ID = t.BUSINESS_ID
and ACT_NAME = t.ACT_NAME
and OPERATE_TYPE = t.OPERATE_TYPE
and OPERATE_TIME > t.OPERATE_TIME );
-- 处理数据1：存在且时戳较大的，需要更新回去
update leodb.t_mdfp_bpm_audit_first as d
join leodb.tmp_mdfp_bpm_audit_history as t on d.BUSINESS_ID = t.BUSINESS_ID and t.OPERATE_TIME > d.OPERATE_TIME
set d.OPERATE_TIME = t.OPERATE_TIME, d.USER_NAME = t.USER_NAME;
-- 处理数据2：不存在的，直接插入
insert into leodb.t_mdfp_bpm_audit_first( BUSINESS_ID, OPERATE_TIME, USER_NAME)
select BUSINESS_ID, OPERATE_TIME, USER_NAME
from leodb.tmp_mdfp_bpm_audit_history as t
where not exists(select 1 from leodb.t_mdfp_bpm_audit_first
where BUSINESS_ID = t.BUSINESS_ID);
-- 释放临时表
drop table leodb.tmp_mdfp_bpm_audit_history;
commit; -- 提交事务
end;
-- 测试，由于以上存储过程，结合了同步周期考虑增量查询，此处入参写大一点，实现全量数据初始化
call leodb.p_job_task_first_audit(10000);
# 3.3、TASK3：预算存物理表
drop PROCEDURE if exists leodb.p_job_task_budget;
create PROCEDURE leodb.p_job_task_budget()
begin
start transaction; -- 开始事务
-- 建表
create table if not exists leodb.t_budget_data
(
FYEAR int,
FMONTH int,
FDATE date,
FYEARMONTH varchar(10),
FBG_DEPT_CODE varchar(50),
FBG_ACCOUNT_CODE varchar(50),
FPROJECT_CODE varchar(50),
FINDUSTRY_CODE varchar(50),
BUDGET_AMOUNT decimal(16,6), -- 预算金额
OCCUPIED_AMOUNT decimal(16,6), -- 占用金额
ACTUAL_AMOUNT decimal(16,6), -- 发生金额
AVAILABLE_AMOUNT decimal(16,6), -- 可用金额
primary key(FYEAR, FMONTH, FBG_DEPT_CODE, FBG_ACCOUNT_CODE, FPROJECT_CODE, FINDUSTRY_CODE)
)
ENGINE=InnoDB DEFAULT CHARSET=utf8;
-- 清除数据
delete from leodb.t_budget_data where fyear = 2023;
-- 插入数据
insert into leodb.t_budget_data( FYEAR, FMONTH, FDATE, FYEARMONTH, FBG_DEPT_CODE, FBG_ACCOUNT_CODE,
FPROJECT_CODE, FINDUSTRY_CODE, BUDGET_AMOUNT, OCCUPIED_AMOUNT, ACTUAL_AMOUNT, AVAILABLE_AMOUNT )
select h.FYEAR, h.FMONTH, h.FDATE, FYEARMONTH, h.FBG_DEPT_CODE, h.FBG_ACCOUNT_CODE,
h.FPROJECT_CODE, h.FINDUSTRY_CODE,
h.BUDGET_AMOUNT, h.OCCUPIED_AMOUNT, h.ACTUAL_AMOUNT, h.AVAILABLE_AMOUNT
from leodb.view_mdfp_bm_format as h
where fyear = 2003;
COMMIT; -- 提交事务
end;
3.4 创建JOB
# 4、创建具体JOB
delimiter #
drop event if exists leodb.JOB_RUN_EVERY1HOUR;
create event leodb.JOB_RUN_EVERY1HOUR
on schedule every 1 hour starts timestamp '2023-06-29 00:00:01'
do
begin
-- 1、维度数据、维度映射转存物理表
set @v_id=cast(unix_timestamp() as signed);
call leodb.p_job_log( @v_id , 'JOB_RUN_EVERY1HOUR', 'leodb.p_job_get_data_4_rp', '');
call leodb.p_job_get_data_4_rp();
call leodb.p_job_log( @v_id , 'JOB_RUN_EVERY1HOUR', 'leodb.p_job_get_data_4_rp', '维度数据转储成功');
-- 2、审批步骤初审人另存物理表
set @v_id=cast(unix_timestamp() as signed);
call leodb.p_job_log( @v_id , 'JOB_RUN_EVERY1HOUR', 'leodb.p_job_task_first_audit(100)', '');
call leodb.p_job_task_first_audit(100); -- 100分钟
call leodb.p_job_log( @v_id , 'JOB_RUN_EVERY1HOUR', 'leodb.p_job_task_first_audit(100)', '初审人员转储成功');
-- 3、预算另存物理表
set @v_id=cast(unix_timestamp() as signed);
call leodb.p_job_log( @v_id , 'JOB_RUN_EVERY1HOUR','leodb.p_job_task_budget', '');
call leodb.p_job_task_budget(); -- 100分钟
call leodb.p_job_log( @v_id , 'JOB_RUN_EVERY1HOUR','leodb.p_job_task_budget', '预算数据转储成功');
end#
delimiter ;
3.5 JOB的维护及查看
# 5、JOB维护
-- 5.1、停止
ALTER EVENT leodb.JOB_RUN_EVERY1HOUR DISABLE;
-- 5.2、开启
ALTER EVENT leodb.JOB_RUN_EVERY1HOUR enable;
-- 5.3、查看状态
select * from mysql.event;
4、总结
通过以上方式的中间数据处理，可以显著提升报表查询效率。
在实际项目中，对数据实时性要求不高的场景，可以使用该方案；对于实时性要求较高的场景，也可以将数据进行分段处理，例如对于历史数据可以先行物化，仅对当前数据进行实时查询，再两者组合来加速全量数据的查询，应用得当也可以有不错的速度提升。
原创文章，转载请注明来源-X档案
