Skip to content

毛树良 / VVAS-DataCenter-DBWriter

Go to a project
Commit acc742b5 by 毛树良
Options

[chg]：全天分析后统计数据

1 parent 700ecbaf
Inline Side-by-side
Showing 1 changed file with 149 additions and 66 deletions
src/main/java/com/viontech/keliu/consumer/MallDataStatisticsConsumer.java
...@@ -4,19 +4,20 @@ import cn.hutool.core.thread.ThreadUtil; ...@@ -4,19 +4,20 @@ import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.http.HttpUtil; import cn.hutool.http.HttpUtil;
import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.ObjectMapper;
import com.viontech.keliu.constants.KafkaConstants; import com.viontech.keliu.constants.KafkaConstants;
import com.viontech.keliu.entity.FaceDataContent;
import com.viontech.keliu.entity.MallDataStatisticsInfo; import com.viontech.keliu.entity.MallDataStatisticsInfo;
import com.viontech.keliu.utils.DateUtil; import com.viontech.keliu.utils.DateUtil;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value; import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener; import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils; import org.springframework.util.CollectionUtils;
import java.util.ArrayList; import java.util.HashMap;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
...@@ -32,19 +33,13 @@ public class MallDataStatisticsConsumer { ...@@ -32,19 +33,13 @@ public class MallDataStatisticsConsumer {
@Value("${vion.consumer.storeDataStatistics.reidUrl:}") @Value("${vion.consumer.storeDataStatistics.reidUrl:}")
private String reidUrl; private String reidUrl;
@Value("${vion.consumer.storeDataStatistics.zoneNum:true}") private ExecutorService executorService = ThreadUtil.newFixedExecutor(10, 1024, "dataStatistics-", true);
private boolean zoneNumEnable;
@Value("${vion.consumer.storeDataStatistics.payEvent:false}")
private boolean payEventEnable;
private ExecutorService executorService = ThreadUtil.newFixedExecutor(1, 1024, "dataStatistics-", true);
@KafkaListener(topics = KafkaConstants.STORE_DATA_STATISTICS_TOPIC @KafkaListener(topics = KafkaConstants.STORE_DATA_STATISTICS_TOPIC
, autoStartup = "${vion.consumer.storeDataStatistics.autoStartup:false}" , autoStartup = "${vion.consumer.storeDataStatistics.autoStartup:false}"
, groupId = "StoreDataStiatistics" , groupId = "StoreDataStiatistics"
, concurrency = "${vion.consumer.storeDataStatistics.concurrency:1}") , concurrency = "${vion.consumer.storeDataStatistics.concurrency:1}")
public void consumerMallZoneMinuteCount(List<ConsumerRecord<String, String>> recordList, Acknowledgment ack) { public void consumerMallZoneMinuteCount(List<ConsumerRecord<String, String>> recordList, Consumer<?, ?> consumer) {
if (CollectionUtils.isEmpty(recordList)) { if (CollectionUtils.isEmpty(recordList)) {
return; return;
} }
...@@ -55,8 +50,6 @@ public class MallDataStatisticsConsumer { ...@@ -55,8 +50,6 @@ public class MallDataStatisticsConsumer {
long startTime = System.currentTimeMillis(); long startTime = System.currentTimeMillis();
List<ConsumerRecord<String, String>> recordValues = entry.getValue(); List<ConsumerRecord<String, String>> recordValues = entry.getValue();
if (!CollectionUtils.isEmpty(recordValues)) { if (!CollectionUtils.isEmpty(recordValues)) {
ConsumerRecord<String, String> lastRecord = recordValues.get(recordValues.size() - 1);
List<FaceDataContent> faceDataList = new ArrayList<>();
for (ConsumerRecord<String, String> consumerRecord : recordValues) { for (ConsumerRecord<String, String> consumerRecord : recordValues) {
try { try {
MallDataStatisticsInfo mallDataStatisticsInfo = objectMapper.readValue(consumerRecord.value(), MallDataStatisticsInfo.class); MallDataStatisticsInfo mallDataStatisticsInfo = objectMapper.readValue(consumerRecord.value(), MallDataStatisticsInfo.class);
...@@ -64,7 +57,13 @@ public class MallDataStatisticsConsumer { ...@@ -64,7 +57,13 @@ public class MallDataStatisticsConsumer {
log.info("处理storeDataStatisticsInfo,mallId:{},countDate:{}, 开始", mallDataStatisticsInfo.getMallId(), mallDataStatisticsInfo.getCountDate()); log.info("处理storeDataStatisticsInfo,mallId:{},countDate:{}, 开始", mallDataStatisticsInfo.getMallId(), mallDataStatisticsInfo.getCountDate());
execDataStatistics(mallDataStatisticsInfo); execDataStatistics(mallDataStatisticsInfo);
}); });
// execDataStatistics(mallDataStatisticsInfo); // 提交Offset
Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
offsets.put(
new TopicPartition(consumerRecord.topic(), consumerRecord.partition()),
new OffsetAndMetadata(consumerRecord.offset() + 1) // 提交下一条偏移量
);
consumer.commitSync(offsets);
} catch (Exception ee) { } catch (Exception ee) {
log.error("处理storeDataStatisticsInfo[{}], JsonDeserializerThrowable={}", entry.getKey(), ee.getMessage(), ee); log.error("处理storeDataStatisticsInfo[{}], JsonDeserializerThrowable={}", entry.getKey(), ee.getMessage(), ee);
} }
...@@ -78,7 +77,6 @@ public class MallDataStatisticsConsumer { ...@@ -78,7 +77,6 @@ public class MallDataStatisticsConsumer {
} catch (Throwable exx) { } catch (Throwable exx) {
log.error("处理mallDataStatisticsInfo.Throwable={}", exx.getMessage(), exx); log.error("处理mallDataStatisticsInfo.Throwable={}", exx.getMessage(), exx);
} }
ack.acknowledge();
} }
private void execDataStatistics(MallDataStatisticsInfo mallDataStatisticsInfo) { private void execDataStatistics(MallDataStatisticsInfo mallDataStatisticsInfo) {
...@@ -93,15 +91,8 @@ public class MallDataStatisticsConsumer { ...@@ -93,15 +91,8 @@ public class MallDataStatisticsConsumer {
} }
log.info("开始执行mallId:{} ,countDate:{} 统计任务", mallDataStatisticsInfo.getMallId(), mallDataStatisticsInfo.getCountDate()); log.info("开始执行mallId:{} ,countDate:{} 统计任务", mallDataStatisticsInfo.getMallId(), mallDataStatisticsInfo.getCountDate());
String date = DateUtil.format(mallDataStatisticsInfo.getCountDate(), "yyyy-MM-dd"); String date = DateUtil.format(mallDataStatisticsInfo.getCountDate(), "yyyy-MM-dd");
String url = reidUrl + "/reid/tool/zoneAssociation?mallIds=" + mallDataStatisticsInfo.getMallId() + "&startDate=" + date + "&endDate=" + date;
try {
String result = HttpUtil.get(url);
log.info("调用mallId:{} 店铺关联 执行结果:{}", mallDataStatisticsInfo.getMallId(), result);
} catch (Exception e) {
log.error("店铺关联计算异常:{}", url, e);
}
url = reidUrl + "/reid/tool/residenceByMall?mallId=" + mallDataStatisticsInfo.getMallId() + "&countDate=" + date; String url = reidUrl + "/reid/tool/residenceByMall?mallId=" + mallDataStatisticsInfo.getMallId() + "&countDate=" + date;
try { try {
String result = HttpUtil.get(url); String result = HttpUtil.get(url);
log.info("调用mallId:{} 平均滞留时间 执行结果:{}", mallDataStatisticsInfo.getMallId(), result); log.info("调用mallId:{} 平均滞留时间 执行结果:{}", mallDataStatisticsInfo.getMallId(), result);
...@@ -117,96 +108,188 @@ public class MallDataStatisticsConsumer { ...@@ -117,96 +108,188 @@ public class MallDataStatisticsConsumer {
log.error("滞留时间计算异常:{}", url, e); log.error("滞留时间计算异常:{}", url, e);
} }
url = reidUrl + "/reid/tool/mallResidence?mallIds=" + mallDataStatisticsInfo.getMallId() + "&startDate=" + date + "&endDate=" + date; url = reidUrl + "/reid/tool/acrossCustomer/reCall?mallIds=" + mallDataStatisticsInfo.getMallId() + "&startDate=" + date + "&endDate=" + date;
try { try {
String result = HttpUtil.get(url); String result = HttpUtil.get(url);
log.info("调用mallId:{} 店铺平均滞留时间 执行结果:{}", mallDataStatisticsInfo.getMallId(), result); log.info("调用mallId:{} 穿堂客流统计 执行结果:{}", mallDataStatisticsInfo.getMallId(), result);
} catch (Exception e) { } catch (Exception e) {
log.error("店铺平均滞留时间计算异常:{}", url, e); log.error("穿堂客流计算异常:{}", url, e);
} }
url = reidUrl + "/reid/tool/zoneFaceResidence?mallIds=" + mallDataStatisticsInfo.getMallId() + "&startDate=" + date + "&endDate=" + date; url = reidUrl + "/reid/tool/dataStat/reCallOne?mallIds=" + mallDataStatisticsInfo.getMallId() + "&startDate=" + date + "&endDate=" + date;
try { try {
String result = HttpUtil.get(url); String result = HttpUtil.get(url);
log.info("调用mallId:{} 店铺滞留时间 执行结果:{}", mallDataStatisticsInfo.getMallId(), result); log.info("调用mallId:{} 客流数据重汇 执行结果:{}", mallDataStatisticsInfo.getMallId(), result);
} catch (Exception e) { } catch (Exception e) {
log.error("店铺滞留时间计算异常:{}", url, e); log.error("客流数据重汇异常:{}", url, e);
} }
url = reidUrl + "/reid/tool/zoneAcrossCustomer?mallIds=" + mallDataStatisticsInfo.getMallId() + "&startDate=" + date + "&endDate=" + date; url = reidUrl + "/reid/tool/faceRecognition/aggregation?mallIds=" + mallDataStatisticsInfo.getMallId() + "&startDate=" + date + "&endDate=" + date;
try { try {
String result = HttpUtil.get(url); String result = HttpUtil.get(url);
log.info("调用mallId:{} 穿堂客流 执行结果:{}", mallDataStatisticsInfo.getMallId(), result); log.info("调用mallId:{} 监控点进入人次统计 执行结果:{}", mallDataStatisticsInfo.getMallId(), result);
} catch (Exception e) { } catch (Exception e) {
log.error("穿堂客流计算异常:{}", url, e); log.error("监控点进入人次统计异常:{}", url, e);
} }
url = reidUrl + "/reid/tool/reStatistics?mallIds=" + mallDataStatisticsInfo.getMallId() + "&startDate=" + date + "&endDate=" + date; url = reidUrl + "/reid/tool/gateFaceResidence?mallIds=" + mallDataStatisticsInfo.getMallId() + "&startDate=" + date + "&endDate=" + date;
try { try {
String result = HttpUtil.get(url); String result = HttpUtil.get(url);
log.info("调用mallId:{} 客流数据重绘 执行结果:{}", mallDataStatisticsInfo.getMallId(), result); log.info("调用mallId:{} 区域滞留时间 执行结果:{}", mallDataStatisticsInfo.getMallId(), result);
} catch (Exception e) { } catch (Exception e) {
log.error("客流数据重绘计算异常:{}", url, e); log.error("区域滞留时间异常:{}", url, e);
} }
url = reidUrl + "/reid/tool/zoneCustomerOverlap?mallIds=" + mallDataStatisticsInfo.getMallId() + "&startDate=" + date + "&endDate=" + date; url = reidUrl + "/reid/tool/cashier?mallIds=" + mallDataStatisticsInfo.getMallId() + "&startDate=" + date + "&endDate=" + date;
try { try {
String result = HttpUtil.get(url); String result = HttpUtil.get(url);
log.info("调用mallId:{} 交叉客群 执行结果:{}", mallDataStatisticsInfo.getMallId(), result); log.info("调用mallId:{} 收银区域客流 执行结果:{}", mallDataStatisticsInfo.getMallId(), result);
} catch (Exception e) { } catch (Exception e) {
log.error("交叉客群计算异常:{}", url, e); log.error("收银区域客流异常:{}", url, e);
} }
url = reidUrl + "/reid/tool/passengerSource?mallIds=" + mallDataStatisticsInfo.getMallId() + "&startDate=" + date + "&endDate=" + date; url = reidUrl + "/reid/tool/customerGroup/reCall?mallIds=" + mallDataStatisticsInfo.getMallId() + "&startDate=" + date + "&endDate=" + date;
try { try {
String result = HttpUtil.get(url); String result = HttpUtil.get(url);
log.info("调用mallId:{} 客流来源 执行结果:{}", mallDataStatisticsInfo.getMallId(), result); log.info("调用mallId:{} 顾客组数 执行结果:{}", mallDataStatisticsInfo.getMallId(), result);
} catch (Exception e) { } catch (Exception e) {
log.error("客流来源计算异常:{}", url, e); log.error("顾客组数计算异常:{}", url, e);
} }
url = reidUrl + "/reid/tool/zoneDeepShopping?mallIds=" + mallDataStatisticsInfo.getMallId() + "&startDate=" + date + "&endDate=" + date; url = reidUrl + "/reid/tool/personRecord/reCall?mallIds=" + mallDataStatisticsInfo.getMallId() + "&startDate=" + date + "&endDate=" + date;
try { try {
String result = HttpUtil.get(url); String result = HttpUtil.get(url);
log.info("调用mallId:{} 深逛数据 执行结果:{}", mallDataStatisticsInfo.getMallId(), result); log.info("调用mallId:{} 接待分析统计 执行结果:{}", mallDataStatisticsInfo.getMallId(), result);
} catch (Exception e) { } catch (Exception e) {
log.error("深逛数据计算异常:{}", url, e); log.error("接待分析统计异常:{}", url, e);
} }
url = reidUrl + "/reid/tool/customerGroup/reCall?mallIds=" + mallDataStatisticsInfo.getMallId() + "&startDate=" + date + "&endDate=" + date; url = reidUrl + "/reid/tool/deepNum?mallIds=" + mallDataStatisticsInfo.getMallId() + "&startDate=" + date + "&endDate=" + date;
try { try {
String result = HttpUtil.get(url); String result = HttpUtil.get(url);
log.info("调用mallId:{} 顾客组数 执行结果:{}", mallDataStatisticsInfo.getMallId(), result); log.info("调用mallId:{} 深逛人数统计 执行结果:{}", mallDataStatisticsInfo.getMallId(), result);
} catch (Exception e) { } catch (Exception e) {
log.error("顾客组数计算异常:{}", url, e); log.error("深逛人数统计异常:{}", url, e);
} }
url = reidUrl + "/reid/tool/archive/get?mallId=" + mallDataStatisticsInfo.getMallId() + "&countDate=" + date; url = reidUrl + "/reid/tool/heatmap/merge?mallIds=" + mallDataStatisticsInfo.getMallId() + "&startDate=" + date + "&endDate=" + date;
try { try {
String result = HttpUtil.get(url); String result = HttpUtil.get(url);
log.info("调用mallId:{} 顾客档案 执行结果:{}", mallDataStatisticsInfo.getMallId(), result); log.info("调用mallId:{} 每日热力数据聚合 执行结果:{}", mallDataStatisticsInfo.getMallId(), result);
} catch (Exception e) { } catch (Exception e) {
log.error("顾客档案计算异常:{}", url, e); log.error("每日热力数据聚合异常:{}", url, e);
} }
if (zoneNumEnable) { url = reidUrl + "/reid/tool/dataMonitor/reCall?mallIds=" + mallDataStatisticsInfo.getMallId() + "&startDate=" + date + "&endDate=" + date;
url = reidUrl + "/reid/tool/mallGateFromTo?mallId=" + mallDataStatisticsInfo.getMallId() + "&countDate=" + date; try {
try { String result = HttpUtil.get(url);
String result = HttpUtil.get(url); log.info("调用mallId:{} 数据运维 执行结果:{}", mallDataStatisticsInfo.getMallId(), result);
log.info("调用mallId:{} 顾客导流 执行结果:{}", mallDataStatisticsInfo.getMallId(), result); } catch (Exception e) {
} catch (Exception e) { log.error("数据运维异常:{}", url, e);
log.error("顾客导流计算异常:{}", url, e);
}
} }
if (payEventEnable) { url = reidUrl + "/reid/tool/mallCompare/reCall?mallIds=" + mallDataStatisticsInfo.getMallId() + "&startDate=" + date + "&endDate=" + date;
url = reidUrl + "/reid/tool/payEvent?mallId=" + mallDataStatisticsInfo.getMallId() + "&countDate=" + date; try {
try { String result = HttpUtil.get(url);
String result = HttpUtil.get(url); log.info("调用mallId:{} 跨店比对 执行结果:{}", mallDataStatisticsInfo.getMallId(), result);
log.info("调用mallId:{} 收银事件 执行结果:{}", mallDataStatisticsInfo.getMallId(), result); } catch (Exception e) {
} catch (Exception e) { log.error("跨店比对异常:{}", url, e);
log.error("收银事件计算异常:{}", url, e); }
}
url = reidUrl + "/reid/tool/kafkaPersonReception/resend?mallIds=" + mallDataStatisticsInfo.getMallId() + "&startDate=" + date + "&endDate=" + date;
try {
String result = HttpUtil.get(url);
log.info("调用mallId:{} 接待数据发送kafka 执行结果:{}", mallDataStatisticsInfo.getMallId(), result);
} catch (Exception e) {
log.error("接待数据发送kafka异常:{}", url, e);
}
url = reidUrl + "/reid/tool/restaurant/countData?mallIds=" + mallDataStatisticsInfo.getMallId() + "&startDate=" + date + "&endDate=" + date;
try {
String result = HttpUtil.get(url);
log.info("调用mallId:{} 餐厅排队指标统计 执行结果:{}", mallDataStatisticsInfo.getMallId(), result);
} catch (Exception e) {
log.error("餐厅排队指标统计异常:{}", url, e);
}
url = reidUrl + "/reid/tool/channelProduct/countData?mallIds=" + mallDataStatisticsInfo.getMallId() + "&startDate=" + date + "&endDate=" + date;
try {
String result = HttpUtil.get(url);
log.info("调用mallId:{} 机位数据统计 执行结果:{}", mallDataStatisticsInfo.getMallId(), result);
} catch (Exception e) {
log.error("机位数据统计异常:{}", url, e);
}
url = reidUrl + "/reid/tool/mallProduct/countData?mallIds=" + mallDataStatisticsInfo.getMallId() + "&startDate=" + date + "&endDate=" + date;
try {
String result = HttpUtil.get(url);
log.info("调用mallId:{} 商品数据统计 执行结果:{}", mallDataStatisticsInfo.getMallId(), result);
} catch (Exception e) {
log.error("商品数据统计异常:{}", url, e);
}
url = reidUrl + "/reid/tool/mallProductGate/countData?mallIds=" + mallDataStatisticsInfo.getMallId() + "&startDate=" + date + "&endDate=" + date;
try {
String result = HttpUtil.get(url);
log.info("调用mallId:{} 商品分区数据统计 执行结果:{}", mallDataStatisticsInfo.getMallId(), result);
} catch (Exception e) {
log.error("商品分区数据统计异常:{}", url, e);
}
url = reidUrl + "/reid/tool/gateFlowDirection?mallIds=" + mallDataStatisticsInfo.getMallId() + "&startDate=" + date + "&endDate=" + date;
try {
String result = HttpUtil.get(url);
log.info("调用mallId:{} 门店分区客流流向 执行结果:{}", mallDataStatisticsInfo.getMallId(), result);
} catch (Exception e) {
log.error("门店分区客流流向异常:{}", url, e);
}
url = reidUrl + "/reid/tool/mallBusinessRecord?mallIds=" + mallDataStatisticsInfo.getMallId() + "&startDate=" + date + "&endDate=" + date;
try {
String result = HttpUtil.get(url);
log.info("调用mallId:{} 门店营业时间统计任务 执行结果:{}", mallDataStatisticsInfo.getMallId(), result);
} catch (Exception e) {
log.error("门店营业时间统计任务异常:{}", url, e);
}
url = reidUrl + "/reid/tool/mallResidenceDistribution?mallIds=" + mallDataStatisticsInfo.getMallId() + "&startDate=" + date + "&endDate=" + date;
try {
String result = HttpUtil.get(url);
log.info("调用mallId:{} 门店停留时长分布 执行结果:{}", mallDataStatisticsInfo.getMallId(), result);
} catch (Exception e) {
log.error("门店停留时长分布异常:{}", url, e);
}
url = reidUrl + "/reid/tool/effectVisitors/update?mallIds=" + mallDataStatisticsInfo.getMallId() + "&startDate=" + date + "&endDate=" + date;
try {
String result = HttpUtil.get(url);
log.info("调用mallId:{} 门店有效顾客统计 执行结果:{}", mallDataStatisticsInfo.getMallId(), result);
} catch (Exception e) {
log.error("门店有效顾客统计异常:{}", url, e);
}
url = reidUrl + "/reid/tool/visitorsMovingLing?mallIds=" + mallDataStatisticsInfo.getMallId() + "&startDate=" + date + "&endDate=" + date;
try {
String result = HttpUtil.get(url);
log.info("调用mallId:{} 顾客店内分区轨迹 执行结果:{}", mallDataStatisticsInfo.getMallId(), result);
} catch (Exception e) {
log.error("顾客店内分区轨迹异常:{}", url, e);
}
url = reidUrl + "/reid/tool/fittingVisitors?mallIds=" + mallDataStatisticsInfo.getMallId() + "&startDate=" + date + "&endDate=" + date;
try {
String result = HttpUtil.get(url);
log.info("调用mallId:{} 试衣顾客统计 执行结果:{}", mallDataStatisticsInfo.getMallId(), result);
} catch (Exception e) {
log.error("试衣顾客统计异常:{}", url, e);
}
url = reidUrl + "/reid/tool/attentionPassersby?mallIds=" + mallDataStatisticsInfo.getMallId() + "&startDate=" + date + "&endDate=" + date;
try {
String result = HttpUtil.get(url);
log.info("调用mallId:{} 过店关注人次 执行结果:{}", mallDataStatisticsInfo.getMallId(), result);
} catch (Exception e) {
log.error("过店关注人次异常:{}", url, e);
} }
} }
} }
Markdown is supported
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!