|
|
@@ -0,0 +1,141 @@
|
|
|
+package com.tzld.piaoquan.recommend.server.service.funnel;
|
|
|
+
|
|
|
+import com.google.common.base.Strings;
|
|
|
+import com.tzld.piaoquan.recommend.server.gen.recommend.RecommendRequest;
|
|
|
+import com.tzld.piaoquan.recommend.server.model.RecommendParam;
|
|
|
+import com.tzld.piaoquan.recommend.server.model.Video;
|
|
|
+import com.tzld.piaoquan.recommend.server.service.rank.RankResult;
|
|
|
+import com.tzld.piaoquan.recommend.server.service.recall.RecallResult;
|
|
|
+import com.tzld.piaoquan.recommend.server.util.JSONUtils;
|
|
|
+import org.apache.commons.collections4.CollectionUtils;
|
|
|
+
|
|
|
+import java.util.ArrayList;
|
|
|
+import java.util.HashMap;
|
|
|
+import java.util.HashSet;
|
|
|
+import java.util.LinkedHashMap;
|
|
|
+import java.util.List;
|
|
|
+import java.util.Map;
|
|
|
+import java.util.Set;
|
|
|
+
|
|
|
+/**
|
|
|
+ * 把 4 阶段(raw / filtered / ranked / preview)合并成每 (videoId, pushFrom) 一行。
|
|
|
+ * 下游 BI: group by videoId max(各 flag) 出整体漏斗;group by pushFrom 出分策略漏斗。
|
|
|
+ */
|
|
|
+public class FunnelAggregator {
|
|
|
+
|
|
|
+ private static class Row {
|
|
|
+ boolean inRecall;
|
|
|
+ boolean inFiltered;
|
|
|
+ double recallScore;
|
|
|
+ int strategyRank;
|
|
|
+ }
|
|
|
+
|
|
|
+ public static List<Map<String, String>> build(
|
|
|
+ String traceId,
|
|
|
+ RecommendRequest request,
|
|
|
+ RecommendParam param,
|
|
|
+ Map<String, List<FunnelRawItem>> funnelSink,
|
|
|
+ RecallResult recallResult,
|
|
|
+ RankResult rankResult,
|
|
|
+ List<Video> returnedVideos) {
|
|
|
+
|
|
|
+ // rank 阶段:被 mergeAndRankRovRecall 挑中进打分的视频 id 集合
|
|
|
+ Set<Long> rankedSet = new HashSet<>();
|
|
|
+ if (rankResult != null && rankResult.getCandidateVideoIds() != null) {
|
|
|
+ rankedSet.addAll(rankResult.getCandidateVideoIds());
|
|
|
+ }
|
|
|
+
|
|
|
+ // rank 最终位置 + 最终 score (rankResult.videos 是 rank 截断前的整序列)
|
|
|
+ Map<Long, Integer> rankPosMap = new HashMap<>();
|
|
|
+ Map<Long, Double> rankScoreMap = new HashMap<>();
|
|
|
+ if (rankResult != null && CollectionUtils.isNotEmpty(rankResult.getVideos())) {
|
|
|
+ List<Video> rv = rankResult.getVideos();
|
|
|
+ for (int i = 0; i < rv.size(); i++) {
|
|
|
+ Video v = rv.get(i);
|
|
|
+ rankPosMap.put(v.getVideoId(), i + 1);
|
|
|
+ rankScoreMap.put(v.getVideoId(), v.getSortScore());
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ // preview = 截断后下发
|
|
|
+ Map<Long, Integer> previewPosMap = new HashMap<>();
|
|
|
+ if (CollectionUtils.isNotEmpty(returnedVideos)) {
|
|
|
+ for (int i = 0; i < returnedVideos.size(); i++) {
|
|
|
+ previewPosMap.put(returnedVideos.get(i).getVideoId(), i + 1);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ // (pushFrom -> videoId -> Row)
|
|
|
+ Map<String, Map<Long, Row>> table = new LinkedHashMap<>();
|
|
|
+
|
|
|
+ // raw
|
|
|
+ if (funnelSink != null) {
|
|
|
+ funnelSink.forEach((pushFrom, items) -> {
|
|
|
+ Map<Long, Row> byVid = table.computeIfAbsent(pushFrom, k -> new LinkedHashMap<>());
|
|
|
+ for (FunnelRawItem item : items) {
|
|
|
+ Row r = byVid.computeIfAbsent(item.getVideoId(), k -> new Row());
|
|
|
+ r.inRecall = true;
|
|
|
+ r.recallScore = item.getScore();
|
|
|
+ r.strategyRank = item.getStrategyRank();
|
|
|
+ }
|
|
|
+ });
|
|
|
+ }
|
|
|
+
|
|
|
+ // filtered
|
|
|
+ if (recallResult != null && CollectionUtils.isNotEmpty(recallResult.getData())) {
|
|
|
+ for (RecallResult.RecallData d : recallResult.getData()) {
|
|
|
+ if (CollectionUtils.isEmpty(d.getVideos())) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ String pushFrom = d.getPushFrom();
|
|
|
+ Map<Long, Row> byVid = table.computeIfAbsent(pushFrom, k -> new LinkedHashMap<>());
|
|
|
+ for (int i = 0; i < d.getVideos().size(); i++) {
|
|
|
+ Video v = d.getVideos().get(i);
|
|
|
+ Row r = byVid.computeIfAbsent(v.getVideoId(), k -> new Row());
|
|
|
+ r.inFiltered = true;
|
|
|
+ // 兜底:strategy 不走 FilterService 时 raw 没记,从 Video 补
|
|
|
+ if (!r.inRecall) {
|
|
|
+ r.recallScore = v.getRovScore();
|
|
|
+ r.strategyRank = i;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ List<Map<String, String>> rows = new ArrayList<>();
|
|
|
+ Map<String, String> base = buildBase(traceId, request, param);
|
|
|
+ table.forEach((pushFrom, byVid) -> byVid.forEach((vid, r) -> {
|
|
|
+ Map<String, String> row = new HashMap<>(base);
|
|
|
+ row.put("videoId", String.valueOf(vid));
|
|
|
+ row.put("pushFrom", pushFrom);
|
|
|
+ row.put("recallScore", String.valueOf(r.recallScore));
|
|
|
+ row.put("strategyRank", String.valueOf(r.strategyRank));
|
|
|
+ row.put("inRecall", r.inRecall ? "1" : "0");
|
|
|
+ row.put("inFiltered", r.inFiltered ? "1" : "0");
|
|
|
+ row.put("inRanked", rankedSet.contains(vid) ? "1" : "0");
|
|
|
+ row.put("inPreview", previewPosMap.containsKey(vid) ? "1" : "0");
|
|
|
+ row.put("rankScore", String.valueOf(rankScoreMap.getOrDefault(vid, 0.0)));
|
|
|
+ row.put("rankPos", String.valueOf(rankPosMap.getOrDefault(vid, 0)));
|
|
|
+ row.put("previewPos", String.valueOf(previewPosMap.getOrDefault(vid, 0)));
|
|
|
+ rows.add(row);
|
|
|
+ }));
|
|
|
+ return rows;
|
|
|
+ }
|
|
|
+
|
|
|
+ private static Map<String, String> buildBase(String traceId, RecommendRequest request, RecommendParam param) {
|
|
|
+ Map<String, String> base = new HashMap<>();
|
|
|
+ base.put("traceId", Strings.nullToEmpty(traceId));
|
|
|
+ if (request != null) {
|
|
|
+ base.put("recommendTraceId", Strings.nullToEmpty(request.getRecommendTraceId()));
|
|
|
+ base.put("sessionId", Strings.nullToEmpty(request.getSessionId()));
|
|
|
+ base.put("rootSessionId", Strings.nullToEmpty(request.getRootSessionId()));
|
|
|
+ base.put("mid", Strings.nullToEmpty(request.getMid()));
|
|
|
+ base.put("appType", String.valueOf(request.getAppType()));
|
|
|
+ base.put("newexpgroup", Strings.nullToEmpty(request.getNewExpGroup()));
|
|
|
+ }
|
|
|
+ if (param != null) {
|
|
|
+ base.put("abExpCode", JSONUtils.toJson(param.getAbExpCodes()));
|
|
|
+ }
|
|
|
+ return base;
|
|
|
+ }
|
|
|
+}
|