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

1.增加异常数据量算法

Browse Source 
2.增加稳态异常数据标记
This commit is contained in:
wr
2025-12-05 15:26:45 +08:00
parent 2624d7d495
commit 5afb860467
8 changed files with 362 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -18,6 +18,7 @@ import com.njcn.dataProcess.param.LineCountEvaluateParam;
import com.njcn.dataProcess.pojo.dto.*;
import com.njcn.dataProcess.pojo.po.PqDataVerify;
import com.njcn.dataProcess.pojo.po.PqDataVerifyBak;
import com.njcn.dataProcess.pojo.po.PqDataVerifyCount;
import com.njcn.dataProcess.util.DataCommonUtils;
import com.njcn.dataProcess.util.TimeUtils;
import com.njcn.device.pq.api.LineFeignClient;
@@ -100,6 +101,8 @@ public class DataCleanServiceImpl implements IDataCleanService {
@Resource
private PqDataVerifyNewFeignClient pqDataVerifyNewFeignClient;
@Resource
private PqDataVerifyCountFeignClient pqDataVerifyCountFeignClient;
@Resource
private FileStorageUtil fileStorageUtil;
@Override
@@ -538,7 +541,79 @@ public class DataCleanServiceImpl implements IDataCleanService {
}
if (CollUtil.isNotEmpty(allTimeSet)) {
Map<String,Object> dtoMap = new HashMap<>();
PqDataVerifyCount verifyCount = new PqDataVerifyCount();
verifyCount.setLineId(item.getLineId());
verifyCount.setTime(LocalDate.parse(calculatedParam.getDataDate(), DateTimeFormatter.ofPattern(DatePattern.NORM_DATE_PATTERN)));
verifyCount.setTotalAll(data1.size());
verifyCount.setFlickerAll(data9.size());
List<DataVDto> dataV = data1.stream()
.filter(item1 -> allTimeSet.contains(item1.getMinTime().substring(11, 19)))
.collect(Collectors.toList());
if (CollUtil.isNotEmpty(dataV)) {
verifyCount.setTotal(dataV.size());
dataV.forEach(item1-> item1.setAbnormalFlag(1));
dataVFeignClient.addInfluxDbList(dataV);
}
List<DataIDto> dataI = data2.stream()
.filter(item1 -> allTimeSet.contains(item1.getMinTime().substring(11, 19)))
.collect(Collectors.toList());
if (CollUtil.isNotEmpty(dataI)) {
dataI.forEach(item1-> item1.setAbnormalFlag(1));
dataIFeignClient.addInfluxDbList(dataI);
}
List<DataPltDto> dataPlt = data3.stream()
.filter(item1 -> allTimeSet.contains(item1.getMinTime().substring(11, 19)))
.collect(Collectors.toList());
if (CollUtil.isNotEmpty(dataPlt)) {
dataPlt.forEach(item1-> item1.setAbnormalFlag(1));
dataPltFeignClient.addInfluxDbList(dataPlt);
}
List<DataHarmDto> dataInHarmV = data4.stream()
.filter(item1 -> allTimeSet.contains(item1.getMinTime().substring(11, 19)))
.collect(Collectors.toList());
if (CollUtil.isNotEmpty(dataInHarmV)) {
dataInHarmV.forEach(item1-> item1.setAbnormalFlag(1));
dataInharmVFeignClient.addInfluxDbList(dataInHarmV);
}
List<DataHarmDto> dataHarmRateV = data5.stream()
.filter(item1 -> allTimeSet.contains(item1.getMinTime().substring(11, 19)))
.collect(Collectors.toList());
if (CollUtil.isNotEmpty(dataHarmRateV)) {
dataHarmRateV.forEach(item1-> item1.setAbnormalFlag(1));
dataHarmRateVFeignClient.addInfluxDbList(dataHarmRateV);
}
List<DataPowerPDto> dataHarmPowerP = data6.stream()
.filter(item1 -> allTimeSet.contains(item1.getMinTime().substring(11, 19)))
.collect(Collectors.toList());
if (CollUtil.isNotEmpty(dataHarmPowerP)) {
dataHarmPowerP.forEach(item1-> item1.setAbnormalFlag(1));
dataHarmpowerPFeignClient.addInfluxDbList(dataHarmPowerP);
}
List<DataHarmDto> dataHarmPhasicV = data7.stream()
.filter(item1 -> allTimeSet.contains(item1.getMinTime().substring(11, 19)))
.collect(Collectors.toList());
if (CollUtil.isNotEmpty(dataHarmPhasicV)) {
dataHarmPhasicV.forEach(item1-> item1.setAbnormalFlag(1));
dataHarmphasicVFeignClient.addInfluxDbList(dataHarmPhasicV);
}
List<DataFlucDto> dataFluc = data8.stream()
.filter(item1 -> allTimeSet.contains(item1.getMinTime().substring(11, 19)))
.collect(Collectors.toList());
if (CollUtil.isNotEmpty(dataFluc)) {
dataFluc.forEach(item1-> item1.setAbnormalFlag(1));
dataFlucFeignClient.addInfluxDbList(dataFluc);
}
List<DataFlickerDto> dataFlicker = data9.stream()
.filter(item1 -> allTimeSet.contains(item1.getMinTime().substring(11, 19)))
.collect(Collectors.toList());
if (CollUtil.isNotEmpty(dataFlicker)) {
verifyCount.setFlicker(dataFlicker.size());
dataFlicker.forEach(item1-> item1.setAbnormalFlag(1));
dataFlickerFeignClient.addInfluxDbList(dataFlicker);
}
pqDataVerifyCountFeignClient.insertData(verifyCount);
Map<String,Object> dtoMap = new HashMap<>(1);
dtoMap.put("lineErrorTimes",allTimeSet.size() * item.getTimeInterval());
resultData.add(dtoMap);
}

28
data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/api/PqDataVerifyCountFeignClient.java Normal file
View File

@@ -0,0 +1,28 @@
package com.njcn.dataProcess.api;
import com.njcn.common.pojo.constant.ServerInfo;
import com.njcn.common.pojo.response.HttpResult;
import com.njcn.dataProcess.api.fallback.PqDataVerifyNewFeignClientFallbackFactory;
import com.njcn.dataProcess.pojo.po.PqDataVerifyBak;
import com.njcn.dataProcess.pojo.po.PqDataVerifyCount;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import java.util.List;
/**
* @author xy
* @version 1.0.0
* @date 2025年02月13日 20:11
*/
@FeignClient(value = ServerInfo.PLATFORM_DATA_PROCESSING_BOOT, path = "/pqDataVerifyCount", fallbackFactory = PqDataVerifyNewFeignClientFallbackFactory.class, contextId = "pqDataVerifyNew")
public interface PqDataVerifyCountFeignClient {
@PostMapping("/insertDataBatch")
HttpResult<List<String>> insertDataBatch(@RequestBody List<PqDataVerifyCount> list);
@PostMapping("/insertData")
HttpResult<List<String>> insertData(@RequestBody PqDataVerifyCount pqDataVerifyCount);
}

54
data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/api/fallback/PqDataVerifyCountFeignClientFallbackFactory.java Normal file
View File

@@ -0,0 +1,54 @@
package com.njcn.dataProcess.api.fallback;
import com.njcn.common.pojo.enums.response.CommonResponseEnum;
import com.njcn.common.pojo.exception.BusinessException;
import com.njcn.common.pojo.response.HttpResult;
import com.njcn.dataProcess.api.PqDataVerifyCountFeignClient;
import com.njcn.dataProcess.api.PqDataVerifyNewFeignClient;
import com.njcn.dataProcess.pojo.po.PqDataVerifyBak;
import com.njcn.dataProcess.pojo.po.PqDataVerifyCount;
import com.njcn.dataProcess.util.DataProcessingEnumUtil;
import feign.hystrix.FallbackFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import java.util.List;
/**
* @author xy
* @version 1.0.0
* @date 2025年02月13日 20:13
*/
@Slf4j
@Component
public class PqDataVerifyCountFeignClientFallbackFactory implements FallbackFactory<PqDataVerifyCountFeignClient> {
/**
* 输出远程请求接口异常日志
* @param cause RPC请求异常
*/
@Override
public PqDataVerifyCountFeignClient create(Throwable cause) {
//判断抛出异常是否为解码器抛出的业务异常
Enum<?> exceptionEnum = CommonResponseEnum.SERVICE_FALLBACK;
if(cause.getCause() instanceof BusinessException){
BusinessException businessException = (BusinessException) cause.getCause();
exceptionEnum = DataProcessingEnumUtil.getExceptionEnum(businessException.getResult());
}
Enum<?> finalExceptionEnum = exceptionEnum;
return new PqDataVerifyCountFeignClient() {
@Override
public HttpResult<List<String>> insertDataBatch(List<PqDataVerifyCount> list) {
log.error("{}异常，降级处理，异常为：{}","批量存储清洗的异常数据",cause.toString());
throw new BusinessException(finalExceptionEnum);
}
@Override
public HttpResult<List<String>> insertData(PqDataVerifyCount pqDataVerifyCount) {
log.error("{}异常，降级处理，异常为：{}","单监测点存储清洗的异常数据",cause.toString());
throw new BusinessException(finalExceptionEnum);
}
};
}
}

68
data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/pojo/po/PqDataVerifyCount.java Normal file
View File

@@ -0,0 +1,68 @@
package com.njcn.dataProcess.pojo.po;
import com.baomidou.mybatisplus.annotation.TableField;
import com.baomidou.mybatisplus.annotation.TableName;
import com.fasterxml.jackson.annotation.JsonFormat;
import com.github.jeffreyning.mybatisplus.anno.MppMultiId;
import lombok.Getter;
import lombok.Setter;
import java.io.Serializable;
import java.time.LocalDate;
/**
* <p>
*
* </p>
*
* @author xy
* @since 2025-02-17
*/
@Getter
@Setter
@TableName("pq_data_verify_count")
public class PqDataVerifyCount implements Serializable {
private static final long serialVersionUID = 1L;
/**
* 监测点id
*/
@MppMultiId
@TableField(value = "line_id")
private String lineId;
/**
* 数据时间
*/
@MppMultiId
@TableField(value = "time_id")
@JsonFormat(pattern = "yyyy-MM-dd", timezone = "GMT+8")
private LocalDate time;
/**
* 异常指标数量
*/
@TableField(value = "total")
private Integer total = 0;
/**
* 总指标数量
*/
@TableField(value = "total_all")
private Integer totalAll = 0;
/**
* 异常短时闪变数量
*/
@TableField(value = "flicker")
private Integer flicker = 0;
/**
* 短时闪变总数量
*/
@TableField(value = "flicker_all")
private Integer flickerAll = 0;
}

58
data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/controller/PqDataVerifyCountController.java Normal file
View File

@@ -0,0 +1,58 @@
package com.njcn.dataProcess.controller;
import com.njcn.common.pojo.annotation.OperateInfo;
import com.njcn.common.pojo.constant.OperateType;
import com.njcn.common.pojo.enums.common.LogEnum;
import com.njcn.common.pojo.enums.response.CommonResponseEnum;
import com.njcn.common.pojo.response.HttpResult;
import com.njcn.common.utils.HttpResultUtil;
import com.njcn.dataProcess.pojo.po.PqDataVerifyBak;
import com.njcn.dataProcess.pojo.po.PqDataVerifyCount;
import com.njcn.dataProcess.service.IPqDataVerifyCountService;
import com.njcn.dataProcess.service.IPqDataVerifyNewService;
import com.njcn.web.controller.BaseController;
import io.swagger.annotations.ApiOperation;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;
import java.util.List;
/**
* <p>
* 前端控制器
* </p>
*
* @author xy
* @since 2025-02-17
*/
@RestController
@RequestMapping("/pqDataVerifyCount")
public class PqDataVerifyCountController extends BaseController {
@Resource
private IPqDataVerifyCountService pqDataVerifyNewService;
@OperateInfo(info = LogEnum.BUSINESS_COMMON,operateType = OperateType.ADD)
@PostMapping("/insertDataBatch")
@ApiOperation("存储清洗的异常数据数量")
public HttpResult<List<String>> insertDataBatch(@RequestBody List<PqDataVerifyCount> list) {
String methodDescribe = getMethodDescribe("insertDataBatch");
pqDataVerifyNewService.insertDataBatch(list);
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, null, methodDescribe);
}
@OperateInfo(info = LogEnum.BUSINESS_COMMON,operateType = OperateType.ADD)
@PostMapping("/insertData")
@ApiOperation("存储清洗的异常数据数量")
public HttpResult<List<String>> insertData(@RequestBody PqDataVerifyCount pqDataVerifyCount) {
String methodDescribe = getMethodDescribe("insertData");
pqDataVerifyNewService.insertData(pqDataVerifyCount);
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, null, methodDescribe);
}
}

16
data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/dao/relation/mapper/PqDataVerifyCountMapper.java Normal file
View File

@@ -0,0 +1,16 @@
package com.njcn.dataProcess.dao.relation.mapper;
import com.github.jeffreyning.mybatisplus.base.MppBaseMapper;
import com.njcn.dataProcess.pojo.po.PqDataVerifyCount;
/**
* <p>
* Mapper 接口
* </p>
*
* @author xy
* @since 2025-02-17
*/
public interface PqDataVerifyCountMapper extends MppBaseMapper<PqDataVerifyCount> {
}

28
data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/IPqDataVerifyCountService.java Normal file
View File

@@ -0,0 +1,28 @@
package com.njcn.dataProcess.service;
import com.github.jeffreyning.mybatisplus.service.IMppService;
import com.njcn.dataProcess.pojo.po.PqDataVerifyCount;
import java.util.List;
/**
* <p>
* 服务类
* </p>
*
* @author xy
* @since 2025-02-17
*/
public interface IPqDataVerifyCountService extends IMppService<PqDataVerifyCount> {
/**
* 异常数据插入
* @param list
*/
void insertDataBatch(List<PqDataVerifyCount> list);
/**
* 异常数据插入
*/
void insertData(PqDataVerifyCount pqDataVerifyCount);
}

34
data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/relation/PqDataVerifyCountServiceImpl.java Normal file
View File

@@ -0,0 +1,34 @@
package com.njcn.dataProcess.service.impl.relation;
import com.github.jeffreyning.mybatisplus.service.MppServiceImpl;
import com.njcn.dataProcess.dao.relation.mapper.PqDataVerifyCountMapper;
import com.njcn.dataProcess.dao.relation.mapper.PqDataVerifyNewMapper;
import com.njcn.dataProcess.pojo.po.PqDataVerifyBak;
import com.njcn.dataProcess.pojo.po.PqDataVerifyCount;
import com.njcn.dataProcess.service.IPqDataVerifyCountService;
import com.njcn.dataProcess.service.IPqDataVerifyNewService;
import org.springframework.stereotype.Service;
import java.util.List;
/**
* <p>
* 服务实现类
* </p>
*
* @author xy
* @since 2025-02-17
*/
@Service
public class PqDataVerifyCountServiceImpl extends MppServiceImpl<PqDataVerifyCountMapper, PqDataVerifyCount> implements IPqDataVerifyCountService {
@Override
public void insertDataBatch(List<PqDataVerifyCount> list) {
this.saveOrUpdateBatchByMultiId(list,1000);
}
@Override
public void insertData(PqDataVerifyCount pqDataVerifyCount) {
this.saveOrUpdateByMultiId(pqDataVerifyCount);
}
}