MOD：调试召回和过滤部分

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/framework/recaller/BaseRecaller.java

@@ -195,34 +195,30 @@ public class BaseRecaller<Video> {
         final FilterPipeline<Video> recallFilter = new FilterPipeline<>(this.filterConfig, requestData, user);
 
 
-        final List<Callable<List<RankItem>>> callables = new ArrayList<Callable<List<RankItem>>>();
-        int expectedRecallSum = 0;
+        final List<Callable<List<RankItem>>> callables = new ArrayList<>();
         for (final Map.Entry<Candidate, Queue<Video>> entry : candidateQueueMap.entrySet()) {
-            callables.add(new Callable<List<RankItem>>() {
-                @Override
-                public List<RankItem> call() throws Exception {
-                    List<RankItem> candidateHits = new ArrayList<RankItem>();
-                    final Candidate candidate = entry.getKey();
-                    try {
-                        // 1. filter  TODO 待后续增加自定义filter
-                        Iterable<Entry<Video>> entries = FluentIterable.from(entry.getValue()).filter(new Predicate<Entry<Video>>() {
-                            @Override
-                            public boolean apply(@Nullable Entry<Video> videoEntry) {
-                                return recallFilter.predicate(candidate, videoEntry.item);
-                            }
-                        }).limit(candidate.getCandidateNum());
+            callables.add(() -> {
+                List<RankItem> candidateHits = new ArrayList<>();
+                final Candidate candidate = entry.getKey();
+                try {
+                    // 1. filter  TODO 待后续增加自定义filter
+                    Iterable<Entry<Video>> entries = FluentIterable.from(entry.getValue()).filter(new Predicate<Entry<Video>>() {
+                        @Override
+                        public boolean apply(@Nullable Entry<Video> videoEntry) {
+                            return recallFilter.predicate(candidate, videoEntry.item);
+                        }
+                    }).limit(candidate.getCandidateNum());
 
-                        // 2. toHits
-                        candidateHits.addAll(toHits(entries, candidate));
+                    // 2. toHits
+                    candidateHits.addAll(toHits(entries, candidate));
 
-                        // debug log for tracing
-                        LOGGER.debug("recalled candidate [{}], queue length [{}], expected [{}], hit [{}]",
-                                new Object[]{candidate.getCandidateKey(), entry.getValue().size(), candidate.getCandidateNum(), candidateHits.size()});
-                    } catch (Exception e) {
-                        LOGGER.error("recall filter queue occur error, queue [{}], error: [{}]", candidate.toString(), ExceptionUtils.getFullStackTrace(e));
-                    }
-                    return candidateHits;
+                    // debug log for tracing
+                    LOGGER.debug("recalled candidate [{}], queue length [{}], expected [{}], hit [{}]",
+                            new Object[]{candidate.getCandidateKey(), entry.getValue().size(), candidate.getCandidateNum(), candidateHits.size()});
+                } catch (Exception e) {
+                    LOGGER.error("recall filter queue occur error, queue [{}], error: [{}]", candidate.toString(), ExceptionUtils.getFullStackTrace(e));
                 }
+                return candidateHits;
             });
         }
 

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/framework/recaller/FilterPipeline.java

@@ -19,7 +19,6 @@ public class FilterPipeline<T> {
     private List<AbstractFilter<T>> filters;
     private RecommendRequest requestContext;
     private User user;
-    private int requestIndex;
 
     public FilterPipeline(FilterConfig config,
                         RecommendRequest requestContext,
@@ -28,7 +27,7 @@ public class FilterPipeline<T> {
         this.config = config;
         this.requestContext = requestContext;
         this.user = user;
-        this.filters = new ArrayList<AbstractFilter<T>>();
+        this.filters = new ArrayList<>();
         this.constructFilters(config);
     }
 
@@ -42,8 +41,8 @@ public class FilterPipeline<T> {
             for (FilterConfigInfo filterConfigInfo : config.getFilterConfigInfoList()) {
                 try {
                     AbstractFilter<T> filter = (AbstractFilter) Class.forName(filterConfigInfo.getFilterName())
-                            .getConstructor(FilterConfigInfo.class, RecommendRequest.class, User.class, Integer.class)
-                            .newInstance(filterConfigInfo, this.requestContext, this.user, this.requestIndex);
+                            .getConstructor(FilterConfigInfo.class, RecommendRequest.class, User.class)
+                            .newInstance(filterConfigInfo, this.requestContext, this.user);
 
                     this.filters.add(filter);
                 } catch (Exception e) {

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/framework/recaller/provider/Index.java

@@ -1,27 +1,28 @@
 package com.tzld.piaoquan.recommend.server.framework.recaller.provider;
 
 import com.google.common.reflect.TypeToken;
-import com.tzld.piaoquan.recommend.server.model.Video;
 import com.tzld.piaoquan.recommend.server.util.JSONUtils;
 import lombok.Data;
 import org.apache.commons.lang3.StringUtils;
 
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 
 @Data
 public class Index {
-    private String indexName;
-    private List<IndexEntry> indexEntryList;
+    private String indexName = null;
+    private List<IndexEntry> indexEntryList = new ArrayList<>();
 
 
     // TODO
     // redis中一路召回队列key 的结果 转化到Index中
-    public Index(String index) {
-        if (StringUtils.isBlank(index)) {
+    public Index(String indexName, String indexValue) {
+        if (StringUtils.isBlank(indexName) || StringUtils.isBlank(indexValue)) {
             return;
         }
-        List<List<String>> videoScores = JSONUtils.fromJson(index, new TypeToken<List<List<String>>>() {
+        this.indexName = indexName;
+        List<List<String>> videoScores = JSONUtils.fromJson(indexValue, new TypeToken<List<List<String>>>() {
         }, Collections.emptyList());
 
         videoScores.stream().forEach(itemAndScore -> {
@@ -29,9 +30,11 @@ public class Index {
                     if (itemAndScore.size() >= 2) {
                         indexEntry.setItemId(itemAndScore.get(0));
                         indexEntry.setScore(Double.valueOf(itemAndScore.get(1)));
+                        this.indexEntryList.add(indexEntry);
                     }
                 }
         );
+
     }
 
 }

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/framework/recaller/provider/RedisBackedQueue.java

@@ -71,7 +71,8 @@ public class RedisBackedQueue implements QueueProvider<Video> {
     // redis读取队列内内容
     public Index get(QueueName name) throws Exception {
         String valueJson = client.get(name.toString());
-        Index index = new Index(valueJson);
+        logger.info("RedisBackedQueue get QueueName: [{}], value: [{}]", name, valueJson);
+        Index index = new Index(name.toString(), valueJson);
         return index;
     }
 

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/framework/recaller/provider/RedisBackedQueueWithoutMetaCacheLoader.java

@@ -29,7 +29,8 @@ public class RedisBackedQueueWithoutMetaCacheLoader extends CacheLoader<QueueNam
     @Override
     public Pair<Long, Queue<Video>> load(QueueName name) throws Exception {
         String valueJson = client.get(name.toString());
-        Index index = new Index(valueJson);
+        logger.info("RedisBackedQueueWithoutMetaCacheLoader get QueueName: [{}], value: [{}]", name, valueJson);
+        Index index = new Index(name.toString(), valueJson);
 
         if (null == index || CollectionUtils.isEmpty(index.getIndexEntryList())) {
             logger.error("empty_index_fetch");

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/implement/FlowPoolRecommendPipeline.java

@@ -55,7 +55,7 @@ public class FlowPoolRecommendPipeline {
 
 
         BaseRecaller recaller = new BaseRecaller(queueProvider);
-        List<RankItem> items = recaller.recalling(requestData, userInfo, requestIndex, new ArrayList<Candidate>(candidates.values()));
+        List<RankItem> items = recaller.recalling(requestData, userInfo, requestIndex, new ArrayList<>(candidates.values()));
 
 
         // Step 4: Advance Scoring

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/implement/TopRecommendPipeline.java

@@ -66,20 +66,17 @@ public class TopRecommendPipeline {
         MergeUtils.distributeItemsToMultiQueues(topQueue, items);
         topQueue.merge(recallNum * 3, userInfo, requestData, requestIndex, 0);
 
-        // step 6：多样性融合
-        List<RankItem> mergeItems = topQueue.getItems();
-        MergeUtils.diversityRerank(mergeItems, SimilarityUtils.getIsSameUserTagOrCategoryFunc(), recallNum, 6, 2);
-
-
-        //
-        // timestamp = System.currentTimeMillis();
-        // RankPipeline rankPipeline = getRankPipeline(requestData;
-        // List<RankItem> resultItems = rankPipeline.doRank(requestData, userInfo, requestIndex, items);
-
+        // 多样性融合
+//        List<RankItem> mergeItems = topQueue.getItems();
+//        MergeUtils.diversityRerank(mergeItems, SimilarityUtils.getIsSameUserTagOrCategoryFunc(), recallNum, 6, 2);
 
         // Step 6: Global Rank & subList
         // TODO: Global Rank
 
+        //
+//         timestamp = System.currentTimeMillis();
+//         RankPipeline rankPipeline = getRankPipeline(requestData;
+//         List<RankItem> resultItems = rankPipeline.doRank(requestData, userInfo, requestIndex, items);
 
         return items;
     }

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/implement/candidate/HotCandidateQueue.java

@@ -10,6 +10,8 @@ import com.tzld.piaoquan.recommend.server.gen.recommend.RecommendRequest;
 import java.util.Map;
 
 /**
+ * 对应 SimHotVideo 和 ReturnVideo
+ *
  * @author sunxy
  */
 public class HotCandidateQueue extends IndexCandidateQueue {
@@ -19,7 +21,9 @@ public class HotCandidateQueue extends IndexCandidateQueue {
     }
 
     @Override
-    public int addCandidateKey(Map<String, Candidate> candidates, int recallNum, User user, RecommendRequest requestData, int requestIndex, int expId) {
+    public int addCandidateKey(Map<String, Candidate> candidates, int recallNum, User user,
+                               RecommendRequest requestData, int requestIndex, int expId) {
+        
         return 0;
     }
 }

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/implement/recall/HistoryLongPeriodFilter.java

@@ -1,7 +1,24 @@
 package com.tzld.piaoquan.recommend.server.implement.recall;
 
+import com.tzld.piaoquan.recommend.server.framework.candidiate.Candidate;
+import com.tzld.piaoquan.recommend.server.framework.common.User;
+import com.tzld.piaoquan.recommend.server.framework.recaller.AbstractFilter;
+import com.tzld.piaoquan.recommend.server.framework.recaller.FilterConfigInfo;
+import com.tzld.piaoquan.recommend.server.gen.recommend.RecommendRequest;
+import com.tzld.piaoquan.recommend.server.model.Video;
+
 /**
  * @author sunxy
  */
-public class HistoryLongPeriodFilter {
+public class HistoryLongPeriodFilter extends AbstractFilter<Video> {
+
+    public HistoryLongPeriodFilter(FilterConfigInfo filterConfigInfo, RecommendRequest requestContext, User user) {
+        super(filterConfigInfo, requestContext, user);
+    }
+
+    @Override
+    public boolean predicate(Candidate candidate, Video aLong) {
+
+        return false;
+    }
 }

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/implement/recall/RiskVideoFilter.java

@@ -30,11 +30,11 @@ public class RiskVideoFilter extends AbstractFilter<Video> {
 
     public boolean isRiskVideo(Video video) {
 
-     // TODO 风险过滤标签加载video中
-     //
-
-
+        // TODO 风险过滤标签加载video中
+        //
+        return false;
 
+    }
 
 
 

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/recall/strategy/AbstractFlowPoolWithScoreRecallStrategy.java

@@ -6,7 +6,6 @@ import com.tzld.piaoquan.recommend.server.service.filter.FlowPoolWithScoreFilter
 import com.tzld.piaoquan.recommend.server.service.recall.FilterParamFactory;
 import com.tzld.piaoquan.recommend.server.service.recall.RecallParam;
 import com.tzld.piaoquan.recommend.server.service.recall.RecallStrategy;
-import com.tzld.piaoquan.recommend.server.util.JSONUtils;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.collections4.CollectionUtils;
 import org.apache.commons.lang3.math.NumberUtils;

recommend-server-service/src/main/resources/filter_config.conf

@@ -1,7 +1,7 @@
 recall-config = {
   filter-config = {
     history-filter-config = {
-      filter-name = "com.tzld.piaoquan.recommend.server.implement.recaller.HistoryLongPeriodFilter"
+      filter-name = "com.tzld.piaoquan.recommend.server.implement.recall.HistoryLongPeriodFilter"
     }
   }
 }