阿里云dataphin平台StarRocks版本 3.3
在dataworks上maxcompute格式运行100s左右，迁移到dataphin之后运行500s，表数据量在360W行左右。
starrocksSQL：
CREATE TABLE IF NOT EXISTS a_qa_infact_sample_assayvalue (
-- 排序键放最前面（StarRocks 要求 DUPLICATE KEY 列必须为表的前 N 列）
`sampleid` varchar(100) COMMENT '质检批号',
`ds` date COMMENT '数据日期（分区字段）',
-- 以下为业务字段
`infact_date` varchar(50) COMMENT '进厂日期',
`samplelotno` varchar(100) COMMENT '试样编号',
`carriertype` varchar(50) COMMENT '运输方式',
`planlistno` varchar(100) COMMENT '业务号',
`pono` varchar(100) COMMENT '合同号',
`fromcode` varchar(50) COMMENT '厂商编号',
`fromdesc` varchar(200) COMMENT '厂商',
`mtrlno` varchar(50) COMMENT '物料编码',
`mtrlname` varchar(200) COMMENT '物料名称',
`areacode` varchar(50) COMMENT '区域',
`netweight` DECIMAL(18,4) COMMENT '净重',
`deductionwater` DECIMAL(18,4) COMMENT '扣水',
`deductionimpurity` DECIMAL(18,4) COMMENT '扣杂',
`deductionstive` DECIMAL(18,4) COMMENT '扣末',
`deductionpiece` DECIMAL(18,4) COMMENT '扣块',
`deduction_all` DECIMAL(18,4) COMMENT '总扣重',
`car_num` DECIMAL(18,4) COMMENT '批次车数',
`element_code` varchar(50) COMMENT '检验指标代码',
`element_name` varchar(100) COMMENT '检验指标名称',
`assayvalue` DECIMAL(18,4) COMMENT '检验结果',
`sendtime` varchar(50) COMMENT '发送时间'
)
ENGINE = OLAP
DUPLICATE KEY (sampleid, ds)
COMMENT ‘外购进厂批次质检结果信息’
PARTITION BY date_trunc(‘day’, ds)
DISTRIBUTED BY HASH(sampleid)
PROPERTIES (
'partition_live_number' = '1095',
'replication_num' = '3',
'compression' = 'LZ4'
);
– ============================================================
– INSERT
– ============================================================
– StarRocks SQL（改写后：先聚合右表再 JOIN，避免 JOIN 阶段数据膨胀）
–********************************************************************–
INSERT OVERWRITE a_qa_infact_sample_assayvalue
(sampleid, ds, infact_date, samplelotno, carriertype, planlistno, pono,
fromcode, fromdesc, mtrlno, mtrlname, areacode, netweight,
deductionwater, deductionimpurity, deductionstive, deductionpiece,
deduction_all, car_num, element_code, element_name, assayvalue, sendtime)
SELECT
t.sampleid
,STR_TO_DATE('${bizdate}', '%Y%m%d') AS ds
,t.infact_date
,t.samplelotno
,t.carriertype
,t.planlistno
,t.pono
,t.fromcode
,t.fromdesc
,t.mtrlno
,t.mtrlname
,t.areacode
,t.netweight
,t.deductionwater
,t.deductionimpurity
,t.deductionstive
,t.deductionpiece
,t.deduction_all
,t.car_num
,a.element_code
,a.element_name
,a.assayvalue
,a.sendtime
FROM mc_cdm.d_fact_qa_infact_sample t
LEFT JOIN (
-- ★ 子查询：先按维度预聚合，每个 (sampleid, samplelotno, mtrlno, 指标) 只有一行
-- 这样 JOIN 时不会产生中间数据膨胀，性能有质的提升
SELECT
sampleid
,samplelotno
,mtrlno
,element_code
,element_name
,MAX(assayvalue) AS assayvalue
,MAX(sendtime) AS sendtime
FROM mc_cdm.d_fact_qa_infact_sample_assayvalue
WHERE ds = '${bizdate}'
GROUP BY sampleid
,samplelotno
,mtrlno
,element_code
,element_name
) a
ON t.sampleid = a.sampleid
AND t.samplelotno = a.samplelotno
AND t.mtrlno = a.mtrlno
WHERE t.ds = ‘${bizdate}’
下面是原平台SQL
CREATE TABLE IF NOT EXISTS a_qa_infact_sample_assayvalue
(
infact_date STRING COMMENT ‘进厂日期’
,sampleid STRING COMMENT ‘质检批号’
,samplelotno STRING COMMENT ‘试样编号’
,carriertype STRING COMMENT ‘运输方式’
,planlistno STRING COMMENT ‘业务号’
,pono STRING COMMENT ‘合同号’
,fromcode STRING COMMENT ‘厂商编号’
,fromdesc STRING COMMENT ‘厂商’
,mtrlno STRING COMMENT ‘物料编码’
,mtrlname STRING COMMENT ‘物料名称’
,areacode STRING COMMENT ‘区域’
,netweight DECIMAL COMMENT ‘净重’
,deductionwater DECIMAL COMMENT ‘扣水’
,deductionimpurity DECIMAL COMMENT ‘扣杂’
,deductionstive DECIMAL COMMENT ‘扣末’
,deductionpiece DECIMAL COMMENT ‘扣块’
,deduction_all DECIMAL COMMENT ‘总扣重’
,car_num DECIMAL COMMENT ‘批次车数’
,element_code STRING COMMENT ‘检验指标代码’
,element_name STRING COMMENT ‘检验指标名称’
,assayvalue DECIMAL COMMENT ‘检验结果’
,sendtime STRING COMMENT ‘发送时间’
)
COMMENT ‘外购进厂批次质检结果信息’
PARTITIONED BY
(
ds STRING
)
;
INSERT OVERWRITE TABLE a_qa_infact_sample_assayvalue PARTITION(ds=’${bdp.system.bizdate}’)
SELECT t.infact_date
,t.sampleid
,t.samplelotno
,t.carriertype
,t.planlistno
,t.pono
,t.fromcode
,t.fromdesc
,t.mtrlno
,t.mtrlname
,t.areacode
,t.netweight
,t.deductionwater
,t.deductionimpurity
,t.deductionstive
,t.deductionpiece
,t.deduction_all
,t.car_num
,a.element_code
,a.element_name
,max(a.assayvalue) as assayvalue
,max(a.sendtime) as sendtime
FROM cdm.d_fact_qa_infact_sample t
LEFT JOIN cdm.d_fact_qa_infact_sample_assayvalue a
ON t.sampleid = a.sampleid
AND t.samplelotno = a.samplelotno
AND t.mtrlno = a.mtrlno
AND a.ds = ‘${bdp.system.bizdate}’
WHERE t.ds = ‘${bdp.system.bizdate}’
group by t.infact_date
,t.sampleid
,t.samplelotno
,t.carriertype
,t.planlistno
,t.pono
,t.fromcode
,t.fromdesc
,t.mtrlno
,t.mtrlname
,t.areacode
,t.netweight
,t.deductionwater
,t.deductionimpurity
,t.deductionstive
,t.deductionpiece
,t.deduction_all
,t.car_num
,a.element_code
,a.element_name
;
所使用到的两张表建表语句
CREATE TABLE IF NOT EXISTS d_fact_qa_infact_sample_assayvalue (
`sampleid` varchar(20) COMMENT '质检批号',
`ds` date COMMENT '数据日期（分区字段）',
`samplelotno` varchar(30) COMMENT '试样编号',
`mtrlno` varchar(20) COMMENT '物料编码',
`mtrlname` varchar(50) COMMENT '物料名称',
`element_code` varchar(50) COMMENT '检验指标代码',
`element_name` varchar(100) COMMENT '检验指标名称',
`assayvalue` decimal(35,5) COMMENT '检验结果',
`sendtime` varchar(19) COMMENT '发送时间'
)
ENGINE = OLAP
DUPLICATE KEY (sampleid, ds)
COMMENT ‘外购进厂检验结果’
PARTITION BY date_trunc(‘day’, ds)
DISTRIBUTED BY HASH(sampleid)
PROPERTIES (
'partition_live_number' = '1095',
'replication_num' = '3',
'compression' = 'LZ4'
);
CREATE TABLE IF NOT EXISTS d_fact_qa_infact_sample (
`infact_date` varchar(8) COMMENT '进厂日期',
`ds` date COMMENT '数据日期（分区字段）',
`sampleid` varchar(20) COMMENT '质检批号',
`samplelotno` varchar(30) COMMENT '试样编号',
`carriertype` varchar(20) COMMENT '运输方式',
`planlistno` varchar(20) COMMENT '业务号',
`pono` varchar(50) COMMENT '合同号',
`fromcode` varchar(20) COMMENT '厂商编号',
`fromdesc` varchar(50) COMMENT '厂商',
`mtrlno` varchar(20) COMMENT '物料编码',
`mtrlname` varchar(50) COMMENT '物料名称',
`areacode` varchar(10) COMMENT '区域',
`netweight` decimal(30,3) COMMENT '净重',
`deductionwater` decimal(30,3) COMMENT '扣水',
`deductionimpurity` decimal(30,3) COMMENT '扣杂',
`deductionstive` decimal(30,3) COMMENT '扣末',
`deductionpiece` decimal(30,3) COMMENT '扣块',
`deduction_all` decimal(30,3) COMMENT '总扣重',
`car_num` bigint COMMENT '批次车数'
)
ENGINE = OLAP
DUPLICATE KEY (infact_date, ds)
COMMENT ‘外购进厂批次基础信息’
PARTITION BY date_trunc(‘day’, ds)
DISTRIBUTED BY HASH(infact_date)
PROPERTIES (
'partition_live_number' = '1095',
'replication_num' = '3',
'compression' = 'LZ4'
);