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:
wr
2025-03-24 10:54:35 +08:00
parent 65f4066dd4
commit 90a5a27db0
10 changed files with 113 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/influxdb/InfluxdbDataFlickerImpl.java
View File

@@ -144,12 +144,21 @@ public class InfluxdbDataFlickerImpl extends MppServiceImpl<RStatDataFlickerRela
influxQueryWrapper.lt(lineParam.getColumnName(), Double.parseDouble(lineParam.getLt()));
}
List<DataFlicker> result = new ArrayList<>();
quality(result, influxQueryWrapper, lineParam);
List<DataFlicker> list = dataFlickerMapper.selectByQueryWrapper(influxQueryWrapper);
if (CollUtil.isNotEmpty(list)) {
Map<String, List<String>> abnormalTime = lineParam.getAbnormalTime();
if (abnormalTime.containsKey(lineParam.getLineId().get(0))) {
List<String> timeList = abnormalTime.get(lineParam.getLineId().get(0));
//有异常数据,当前监测点自身的异常数据
if (CollectionUtil.isNotEmpty(timeList)) {
result.addAll(list.stream().filter(item -> !timeList.contains(DATE_TIME_FORMATTER.format(item.getTime()))).collect(Collectors.toList()));
}
}
}
return result.size();
}
private void quality(List<DataFlicker> result, InfluxQueryWrapper influxQueryWrapper, LineCountEvaluateParam lineParam) {
List<DataFlicker> list = dataFlickerMapper.selectByQueryWrapper(influxQueryWrapper);
Map<String, List<DataFlicker>> lineMap = list.stream().collect(Collectors.groupingBy(DataFlicker::getLineId));
//有异常数据

72
data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/influxdb/InfluxdbDataVImpl.java
View File

@@ -181,8 +181,18 @@ public class InfluxdbDataVImpl extends MppServiceImpl<RStatDataVRelationMapper,
if (ObjectUtil.isNotEmpty(lineParam.getLt())) {
influxQueryWrapper.lt(lineParam.getColumnName(), Double.parseDouble(lineParam.getLt()));
}
List<DataV> result=new ArrayList<>();
quality(result, influxQueryWrapper,lineParam);
List<DataV> result = new ArrayList<>();
List<DataV> list = dataVMapper.selectByQueryWrapper(influxQueryWrapper);
if (CollUtil.isNotEmpty(list)) {
Map<String, List<String>> abnormalTime = lineParam.getAbnormalTime();
if (abnormalTime.containsKey(lineParam.getLineId().get(0))) {
List<String> timeList = abnormalTime.get(lineParam.getLineId().get(0));
//有异常数据,当前监测点自身的异常数据
if (CollectionUtil.isNotEmpty(timeList)) {
result.addAll(list.stream().filter(item -> !timeList.contains(DATE_TIME_FORMATTER.format(item.getTime()))).collect(Collectors.toList()));
}
}
}
return result.size();
}
@@ -339,7 +349,8 @@ public class InfluxdbDataVImpl extends MppServiceImpl<RStatDataVRelationMapper,
}
@Override
public List<MeasurementCountDTO> getMeasurementCount(List<String> lineIndex, String startTime, String endTime) {
public List<MeasurementCountDTO> getMeasurementCount(List<String> lineIndex, String startTime, String
endTime) {
InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(DataV.class, MeasurementCountDTO.class);
influxQueryWrapper.regular(DataV::getLineId, lineIndex)
.eq(DataV::getValueType, InfluxDbSqlConstant.MAX)
@@ -388,40 +399,43 @@ public class InfluxdbDataVImpl extends MppServiceImpl<RStatDataVRelationMapper,
if (CollUtil.isNotEmpty(lineParam.getPhasicType())) {
influxQueryWrapper.regular(DataV::getPhasicType, lineParam.getPhasicType());
}
quality(result, influxQueryWrapper,lineParam);
quality(result, influxQueryWrapper, lineParam);
return result;
}
private void quality(List<DataV> result,InfluxQueryWrapper influxQueryWrapper, LineCountEvaluateParam lineParam) {
private void quality(List<DataV> result, InfluxQueryWrapper influxQueryWrapper, LineCountEvaluateParam
lineParam) {
List<DataV> list = dataVMapper.selectByQueryWrapper(influxQueryWrapper);
Map<String, List<DataV>> lineMap = list.stream().collect(Collectors.groupingBy(DataV::getLineId));
//有异常数据
Map<String, List<String>> timeMap = lineParam.getAbnormalTime();
if (CollectionUtil.isNotEmpty(timeMap)) {
lineMap.forEach((k, v) -> {
List<String> timeList = timeMap.get(k);
//有异常数据,当前监测点自身的异常数据
if (CollectionUtil.isNotEmpty(timeList)) {
List<DataV> filterList = v.stream().filter(item -> !timeList.contains(DATE_TIME_FORMATTER.format(item.getTime()))).collect(Collectors.toList());
//1.过滤掉异常数据后还有正常数据，则用正常数据计算
if (CollectionUtil.isNotEmpty(filterList)) {
result.addAll(filterList);
if (CollUtil.isNotEmpty(list)) {
Map<String, List<DataV>> lineMap = list.stream().collect(Collectors.groupingBy(DataV::getLineId));
//有异常数据
Map<String, List<String>> timeMap = lineParam.getAbnormalTime();
if (CollectionUtil.isNotEmpty(timeMap)) {
lineMap.forEach((k, v) -> {
List<String> timeList = timeMap.get(k);
//有异常数据,当前监测点自身的异常数据
if (CollectionUtil.isNotEmpty(timeList)) {
List<DataV> filterList = v.stream().filter(item -> !timeList.contains(DATE_TIME_FORMATTER.format(item.getTime()))).collect(Collectors.toList());
//1.过滤掉异常数据后还有正常数据，则用正常数据计算
if (CollectionUtil.isNotEmpty(filterList)) {
result.addAll(filterList);
}
//2.过滤掉异常数据后没有正常数据，则用所有异常数据计算，但是需要标记数据为异常的
else {
v.parallelStream().forEach(item -> item.setQualityFlag("1"));
result.addAll(v);
}
}
//2.过滤掉异常数据后没有正常数据，则用所有异常数据计算，但是需要标记数据为异常的
//没有异常数据，则使用原数据
else {
v.parallelStream().forEach(item -> item.setQualityFlag("1"));
result.addAll(v);
}
}
//没有异常数据，则使用原数据
else {
result.addAll(v);
}
});
}
//没有异常数据，则使用原数据
else {
result.addAll(list);
});
}
//没有异常数据，则使用原数据
else {
result.addAll(list);
}
}
}
}

18
data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/relation/RelationDataHarmRateVImpl.java
View File

@@ -3,6 +3,7 @@ package com.njcn.dataProcess.service.impl.relation;
import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.util.ObjectUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.github.jeffreyning.mybatisplus.service.MppServiceImpl;
import com.njcn.dataProcess.dao.relation.mapper.DataHarmrateVRelationMapper;
@@ -52,13 +53,12 @@ public class RelationDataHarmRateVImpl extends MppServiceImpl<RStatDataHarmRateV
}
@Override
public void batchInsertion(List<DataHarmrateVDTO> dataHarmrateVDTOList) {
int totalCount = dataHarmrateVDTOList.size();
int minSize = Math.min(120, totalCount);
if(totalCount<=0){
if (totalCount <= 0) {
return;
}
List<DataHarmrateV> collect = dataHarmrateVDTOList.stream().map(temp -> {
@@ -87,7 +87,7 @@ public class RelationDataHarmRateVImpl extends MppServiceImpl<RStatDataHarmRateV
@Override
public void addList(List<DataHarmRateVDto> list) {
List<RStatDataHarmRateVD> result = new ArrayList<>();
list.forEach(item->{
list.forEach(item -> {
RStatDataHarmRateVD data = new RStatDataHarmRateVD();
data.setTime(TimeUtils.LocalDataTimeToLocalDate2(item.getTime()));
data.setPhasicType(item.getPhasicType());
@@ -107,18 +107,20 @@ public class RelationDataHarmRateVImpl extends MppServiceImpl<RStatDataHarmRateV
public List<DataHarmDto> getHarmRateVData(LineCountEvaluateParam lineParam) {
List<DataHarmDto> result = new ArrayList<>();
List<RStatDataHarmRateVD> rStatDataHarmRateVDList = dataHarmRateV.list(new LambdaQueryWrapper<RStatDataHarmRateVD>()
.in(RStatDataHarmRateVD::getLineId,lineParam.getLineId())
.in(RStatDataHarmRateVD::getLineId, lineParam.getLineId())
.in(RStatDataHarmRateVD::getValueType, lineParam.getValueType())
.in(RStatDataHarmRateVD::getPhasicType, lineParam.getPhasicType())
.ge(RStatDataHarmRateVD::getTime, lineParam.getStartTime())
.le(RStatDataHarmRateVD::getTime, lineParam.getEndTime())
);
for(RStatDataHarmRateVD rStatDataHarmRateVD : rStatDataHarmRateVDList){
DataHarmDto dto = BeanUtil.copyProperties(rStatDataHarmRateVD,DataHarmDto.class);
dto.setTime(rStatDataHarmRateVD.getTime().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATE_PATTERN)));
for (RStatDataHarmRateVD rStatDataHarmRateVD : rStatDataHarmRateVDList) {
DataHarmDto dto = BeanUtil.copyProperties(rStatDataHarmRateVD, DataHarmDto.class);
if (ObjectUtil.isNotNull(rStatDataHarmRateVD.getTime())) {
dto.setTime(rStatDataHarmRateVD.getTime().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATE_PATTERN)));
}
result.add(dto);
}
return quality(result,lineParam);
return quality(result, lineParam);
}
private List<DataHarmDto> quality(List<DataHarmDto> list, LineCountEvaluateParam lineParam) {

21
data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/relation/RelationDataIImpl.java
View File

@@ -3,6 +3,7 @@ package com.njcn.dataProcess.service.impl.relation;
import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
@@ -49,9 +50,9 @@ public class RelationDataIImpl extends MppServiceImpl<RStatDataIRelationMapper,
@Override
public void batchInsertion(List<DataIDTO> dataIDTOList) {
int totalCount = dataIDTOList.size();
int minSize = Math.min(120, totalCount);
int minSize = Math.min(120, totalCount);
if(totalCount<=0){
if (totalCount <= 0) {
return;
}
List<DataI> collect = dataIDTOList.stream().map(temp -> {
@@ -84,7 +85,7 @@ public class RelationDataIImpl extends MppServiceImpl<RStatDataIRelationMapper,
@Override
public void addList(List<DataIDto> dataIDtoList) {
List<RStatDataID> result = new ArrayList<>();
dataIDtoList.forEach(item->{
dataIDtoList.forEach(item -> {
RStatDataID dataI = new RStatDataID();
dataI.setTime(TimeUtils.LocalDataTimeToLocalDate2(item.getTime()));
dataI.setPhasicType(item.getPhasicType());
@@ -100,7 +101,7 @@ public class RelationDataIImpl extends MppServiceImpl<RStatDataIRelationMapper,
List<DataIDto> result = new ArrayList<>();
QueryWrapper<RStatDataID> queryWrapper = new QueryWrapper<>();
if(StrUtil.isNotBlank(lineParam.getColumnName())){
if (StrUtil.isNotBlank(lineParam.getColumnName())) {
queryWrapper.select(lineParam.getColumnName());
}
queryWrapper.lambda()
@@ -108,15 +109,17 @@ public class RelationDataIImpl extends MppServiceImpl<RStatDataIRelationMapper,
.in(RStatDataID::getPhasicType, lineParam.getPhasicType())
.ge(RStatDataID::getTime, lineParam.getStartTime())
.le(RStatDataID::getTime, lineParam.getEndTime())
.in(RStatDataID::getLineId,lineParam.getLineId());
.in(RStatDataID::getLineId, lineParam.getLineId());
List<RStatDataID> list = this.list(queryWrapper);
for(RStatDataID rStatDataID:list){
DataIDto dto =BeanUtil.copyProperties(rStatDataID,DataIDto.class);
dto.setTime(rStatDataID.getTime().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATE_PATTERN)));
for (RStatDataID rStatDataID : list) {
DataIDto dto = BeanUtil.copyProperties(rStatDataID, DataIDto.class);
if (ObjectUtil.isNotNull(rStatDataID.getTime())) {
dto.setTime(rStatDataID.getTime().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATE_PATTERN)));
}
result.add(dto);
}
return quality(result,lineParam);
return quality(result, lineParam);
}
private List<DataIDto> quality(List<DataIDto> list, LineCountEvaluateParam lineParam) {

19
data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/relation/RelationDataPltImpl.java
View File

@@ -3,6 +3,7 @@ package com.njcn.dataProcess.service.impl.relation;
import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.github.jeffreyning.mybatisplus.service.MppServiceImpl;
@@ -46,9 +47,9 @@ public class RelationDataPltImpl extends MppServiceImpl<RStatDataPltRelationMapp
@Override
public void batchInsertion(List<DataPltDTO> dataPltDTOList) {
int totalCount = dataPltDTOList.size();
int minSize = Math.min(120, totalCount);
int minSize = Math.min(120, totalCount);
if(totalCount<=0){
if (totalCount <= 0) {
return;
}
List<DataPlt> collect = dataPltDTOList.stream().map(temp -> {
@@ -82,7 +83,7 @@ public class RelationDataPltImpl extends MppServiceImpl<RStatDataPltRelationMapp
@Override
public void addList(List<DataPltDto> list) {
List<RStatDataPltD> result = new ArrayList<>();
list.forEach(item->{
list.forEach(item -> {
RStatDataPltD data = new RStatDataPltD();
data.setTime(TimeUtils.LocalDataTimeToLocalDate2(item.getTime()));
data.setPhasicType(item.getPhasicType());
@@ -97,7 +98,7 @@ public class RelationDataPltImpl extends MppServiceImpl<RStatDataPltRelationMapp
public List<DataPltDto> getDataPlt(LineCountEvaluateParam lineParam) {
List<DataPltDto> result = new ArrayList<>();
QueryWrapper<RStatDataPltD> queryWrapper = new QueryWrapper<>();
if(StrUtil.isNotBlank(lineParam.getColumnName())){
if (StrUtil.isNotBlank(lineParam.getColumnName())) {
queryWrapper.select(lineParam.getColumnName());
}
queryWrapper.lambda()
@@ -107,12 +108,14 @@ public class RelationDataPltImpl extends MppServiceImpl<RStatDataPltRelationMapp
.le(RStatDataPltD::getTime, lineParam.getEndTime());
List<RStatDataPltD> rStatDataVDList = this.list(queryWrapper);
for(RStatDataPltD rStatDataPltD : rStatDataVDList){
DataPltDto dto = BeanUtil.copyProperties(rStatDataPltD,DataPltDto.class);
dto.setTime(rStatDataPltD.getTime().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATE_PATTERN)));
for (RStatDataPltD rStatDataPltD : rStatDataVDList) {
DataPltDto dto = BeanUtil.copyProperties(rStatDataPltD, DataPltDto.class);
if (ObjectUtil.isNotNull(rStatDataPltD.getTime())) {
dto.setTime(rStatDataPltD.getTime().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATE_PATTERN)));
}
result.add(dto);
}
return quality(result,lineParam);
return quality(result, lineParam);
}
private List<DataPltDto> quality(List<DataPltDto> list, LineCountEvaluateParam lineParam) {

7
data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/relation/RelationDataVImpl.java
View File

@@ -3,6 +3,7 @@ package com.njcn.dataProcess.service.impl.relation;
import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.github.jeffreyning.mybatisplus.service.MppServiceImpl;
@@ -153,10 +154,12 @@ public class RelationDataVImpl extends MppServiceImpl<RStatDataVRelationMapper,
List<RStatDataVD> rStatDataVDList = this.list(queryWrapper);
for (RStatDataVD rStatDataVD : rStatDataVDList) {
DataVDto dto = BeanUtil.copyProperties(rStatDataVD, DataVDto.class);
dto.setTime(rStatDataVD.getTime().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATE_PATTERN)));
if (ObjectUtil.isNotNull(rStatDataVD.getTime())) {
dto.setTime(rStatDataVD.getTime().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATE_PATTERN)));
}
info.add(dto);
}
return quality(info,lineParam);
return quality(info, lineParam);
}
private List<DataVDto> quality(List<DataVDto> list, LineCountEvaluateParam lineParam) {