Middleware/cn-data-platform
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

数据质量清洗算法优化

Browse Source
This commit is contained in:
xy
2025-05-23 16:18:58 +08:00
parent 412a67f6fd
commit 56aa40211d
68 changed files with 471 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files

109
algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/serviceimpl/line/DataCleanServiceImpl.java
View File

@@ -6,7 +6,7 @@ import com.njcn.algorithm.pojo.bo.CalculatedParam;
import com.njcn.algorithm.service.line.IDataCleanService;
import com.njcn.algorithm.utils.MemorySizeUtil;
import com.njcn.dataProcess.api.*;
import com.njcn.dataProcess.dto.*;
import com.njcn.dataProcess.dto.RmpEventDetailDTO;
import com.njcn.dataProcess.enums.DataCleanEnum;
import com.njcn.dataProcess.param.DataCleanParam;
import com.njcn.dataProcess.param.LineCountEvaluateParam;
@@ -22,7 +22,6 @@ import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.ListUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
@@ -160,105 +159,70 @@ public class DataCleanServiceImpl implements IDataCleanService {
//修改dataFlicker数据
List<DataFlickerDto> dataFlicker = dataFlickerFeignClient.getRawData(param).getData();
if (CollUtil.isNotEmpty(dataFlicker)) {
List<DataFlickerDTO> dataFlickerDTOList = new ArrayList<>();
dataFlicker.forEach(item1->{
DataFlickerDTO dto = new DataFlickerDTO();
BeanUtils.copyProperties(item1,dto);
dto.setTimeid(LocalDateTime.parse(item1.getMinTime(), DATE_TIME_FORMATTER));
dto.setAbnormalFlag(1);
dataFlickerDTOList.add(dto);
item1.setAbnormalFlag(1);
});
dataFlickerFeignClient.batchInsertion(dataFlickerDTOList);
dataFlickerFeignClient.addInfluxDbList(dataFlicker);
log.info("修改dataFlicker数据量:{}", dataFlicker.size());
}
//修改dataFluc数据
List<DataFlucDto> dataFluc = dataFlucFeignClient.getRawData(param).getData();
if (CollUtil.isNotEmpty(dataFluc)) {
List<DataFlucDTO> dataFlucDTOList = new ArrayList<>();
dataFluc.forEach(item1->{
DataFlucDTO dto = new DataFlucDTO();
BeanUtils.copyProperties(item1,dto);
dto.setTimeid(LocalDateTime.parse(item1.getMinTime(), DATE_TIME_FORMATTER));
dto.setAbnormalFlag(1);
dataFlucDTOList.add(dto);
item1.setAbnormalFlag(1);
});
dataFlucFeignClient.batchInsertion(dataFlucDTOList);
dataFlucFeignClient.addInfluxDbList(dataFluc);
log.info("修改dataFluc数据量:{}", dataFluc.size());
}
//修改dataHarmPhasicI数据
List<DataHarmPhasicIDto> dataHarmPhasicI = dataHarmphasicIFeignClient.getRawData(param).getData();
if (CollUtil.isNotEmpty(dataHarmPhasicI)) {
List<DataHarmphasicIDTO> dataHarmphasicIDTOList = new ArrayList<>();
dataHarmPhasicI.forEach(item1->{
DataHarmphasicIDTO dto = new DataHarmphasicIDTO();
BeanUtils.copyProperties(item1,dto);
dto.setTimeid(LocalDateTime.parse(item1.getMinTime(), DATE_TIME_FORMATTER));
dto.setAbnormalFlag(1);
dataHarmphasicIDTOList.add(dto);
List<DataHarmPhasicIDto> dataHarmI = dataHarmphasicIFeignClient.getRawData(param).getData();
if (CollUtil.isNotEmpty(dataHarmI)) {
dataHarmI.forEach(item1->{
item1.setAbnormalFlag(1);
});
dataHarmphasicIFeignClient.batchInsertion(dataHarmphasicIDTOList);
log.info("修改dataHarmPhasicI数据量:{}", dataHarmPhasicI.size());
dataHarmphasicIFeignClient.addInfluxDbList(dataHarmI);
log.info("修改dataHarmI数据量:{}", dataHarmI.size());
}
//修改dataHarmPhasicV数据
List<DataHarmDto> DataHarmV = dataHarmphasicVFeignClient.getRawData(param).getData();
if (CollUtil.isNotEmpty(DataHarmV)) {
List<DataHarmphasicVDTO> dataHarmphasicVDTOList = new ArrayList<>();
DataHarmV.forEach(item1->{
DataHarmphasicVDTO dto = new DataHarmphasicVDTO();
BeanUtils.copyProperties(item1,dto);
dto.setTimeid(LocalDateTime.parse(item1.getMinTime(), DATE_TIME_FORMATTER));
dto.setAbnormalFlag(1);
dataHarmphasicVDTOList.add(dto);
item1.setAbnormalFlag(1);
});
dataHarmphasicVFeignClient.batchInsertion(dataHarmphasicVDTOList);
dataHarmphasicVFeignClient.addInfluxDbList(DataHarmV);
log.info("修改dataHarmPhasicV数据量:{}", DataHarmV.size());
}
//修改dataHarmPowerP数据
List<DataPowerPDto> dataPowerP = dataHarmpowerPFeignClient.getRawData(param).getData();
if (CollUtil.isNotEmpty(dataPowerP)) {
List<DataHarmpowerPDTO> dataHarmpowerPDTOList = new ArrayList<>();
dataPowerP.forEach(item1->{
DataHarmpowerPDTO dto = new DataHarmpowerPDTO();
BeanUtils.copyProperties(item1,dto);
dto.setTimeid(LocalDateTime.parse(item1.getMinTime(), DATE_TIME_FORMATTER));
dto.setAbnormalFlag(1);
dataHarmpowerPDTOList.add(dto);
item1.setAbnormalFlag(1);
});
dataHarmpowerPFeignClient.batchInsertion(dataHarmpowerPDTOList);
log.info("修改 dataHarmPowerP数据量:{}", dataPowerP.size());
dataHarmpowerPFeignClient.addInfluxDbList(dataPowerP);
log.info("dataHarmPowerP数据量:{}", dataPowerP.size());
}
//修改dataHarmPowerQ数据
List<DataHarmPowerQDto> dataHarmPowerQ = dataHarmpowerQFeignClient.getRawData(param).getData();
if (CollUtil.isNotEmpty(dataHarmPowerQ)) {
List<DataHarmpowerQDTO> dataHarmpowerQDTOList = new ArrayList<>();
dataHarmPowerQ.forEach(item1->{
DataHarmpowerQDTO dto = new DataHarmpowerQDTO();
BeanUtils.copyProperties(item1,dto);
dto.setTimeid(LocalDateTime.parse(item1.getMinTime(), DATE_TIME_FORMATTER));
dto.setAbnormalFlag(1);
dataHarmpowerQDTOList.add(dto);
item1.setAbnormalFlag(1);
});
dataHarmpowerQFeignClient.batchInsertion(dataHarmpowerQDTOList);
dataHarmpowerQFeignClient.addInfluxDbList(dataHarmPowerQ);
log.info("修改dataHarmPowerQ数据量:{}", dataHarmPowerQ.size());
}
//修改dataHarmPowerS数据
List<DataHarmPowerSDto> dataHarmPowerS = dataHarmpowerSFeignClient.getRawData(param).getData();
if (CollUtil.isNotEmpty(dataHarmPowerS)) {
List<DataHarmpowerSDTO> dataHarmpowerSDTOList = new ArrayList<>();
dataHarmPowerS.forEach(item1->{
DataHarmpowerSDTO dto = new DataHarmpowerSDTO();
BeanUtils.copyProperties(item1,dto);
dto.setTimeid(LocalDateTime.parse(item1.getMinTime(), DATE_TIME_FORMATTER));
dto.setAbnormalFlag(1);
dataHarmpowerSDTOList.add(dto);
item1.setAbnormalFlag(1);
});
dataHarmpowerSFeignClient.batchInsertion(dataHarmpowerSDTOList);
dataHarmpowerSFeignClient.addInfluxDbList(dataHarmPowerS);
log.info("修改dataHarmPowerS数据量:{}", dataHarmPowerS.size());
}
@@ -275,55 +239,40 @@ public class DataCleanServiceImpl implements IDataCleanService {
//修改dataHarmRateV数据
List<DataHarmDto> DataHarmRateV = dataHarmRateVFeignClient.getRawData(param).getData();
if (CollUtil.isNotEmpty(DataHarmRateV)) {
List<DataHarmrateVDTO> dataHarmrateVDTOList = new ArrayList<>();
DataHarmRateV.forEach(item1->{
DataHarmrateVDTO dto = new DataHarmrateVDTO();
BeanUtils.copyProperties(item1,dto);
dto.setTimeid(LocalDateTime.parse(item1.getMinTime(), DATE_TIME_FORMATTER));
dto.setAbnormalFlag(1);
dataHarmrateVDTOList.add(dto);
item1.setAbnormalFlag(1);
});
dataHarmRateVFeignClient.batchInsertion(dataHarmrateVDTOList);
dataHarmRateVFeignClient.addInfluxDbList(DataHarmRateV);
log.info("修改dataHarmRateV数据量:{}", DataHarmRateV.size());
}
//修改dataInHarmI数据
List<DataHarmRateIDto> dataInHarmI = dataHarmRateIFeignClient.getRawData(param).getData();
List<DataInHarmIDto> dataInHarmI = dataInharmIFeignClient.getRawData(param).getData();
if (CollUtil.isNotEmpty(dataInHarmI)) {
dataInHarmI.forEach(item1->{
item1.setAbnormalFlag(1);
});
dataHarmRateIFeignClient.addInfluxDbList(dataInHarmI);
log.info("修改dataHarmRateI数据量:{}", dataInHarmI.size());
dataInharmIFeignClient.addInfluxDbList(dataInHarmI);
log.info("修改dataInHarmI数据量:{}", dataInHarmI.size());
}
//修改dataInHarmV数据
List<DataHarmDto> dataInHarmV = dataInharmVFeignClient.getRawData(param).getData();
if (CollUtil.isNotEmpty(dataInHarmV)) {
List<DataInharmVDTO> dataInharmVDTOList = new ArrayList<>();
dataInHarmV.forEach(item1->{
DataInharmVDTO dto = new DataInharmVDTO();
BeanUtils.copyProperties(item1,dto);
dto.setTimeid(LocalDateTime.parse(item1.getMinTime(), DATE_TIME_FORMATTER));
dto.setAbnormalFlag(1);
dataInharmVDTOList.add(dto);
item1.setAbnormalFlag(1);
});
dataInharmVFeignClient.batchInsertion(dataInharmVDTOList);
dataInharmVFeignClient.addInfluxDbList(dataInHarmV);
log.info("修改dataInHarmV数据量:{}", dataInHarmV.size());
}
//修改dataPlt数据
List<DataPltDto> dataPlt = dataPltFeignClient.getRawData(param).getData();
if (CollUtil.isNotEmpty(dataPlt)) {
List<DataPltDTO> dataPltDTOList = new ArrayList<>();
dataPlt.forEach(item1->{
DataPltDTO pltDTO = new DataPltDTO();
BeanUtils.copyProperties(item1,pltDTO);
pltDTO.setTimeid(LocalDateTime.parse(item1.getMinTime(), DATE_TIME_FORMATTER));
pltDTO.setAbnormalFlag(1);
dataPltDTOList.add(pltDTO);
item1.setAbnormalFlag(1);
});
dataPltFeignClient.batchInsertion(dataPltDTOList);
dataPltFeignClient.addInfluxDbList(dataPlt);
log.info("修改dataPlt数据量:{}", dataPlt.size());
}
});