|
|
@@ -0,0 +1,159 @@
|
|
|
+package com.tzld.piaoquan.recommend.server.service.score;
|
|
|
+
|
|
|
+
|
|
|
+import com.tzld.piaoquan.recommend.feature.domain.video.base.UserFeature;
|
|
|
+import com.tzld.piaoquan.recommend.server.common.base.RankItem;
|
|
|
+import com.tzld.piaoquan.recommend.server.service.score.model.FMModel;
|
|
|
+import com.tzld.piaoquan.recommend.server.util.JSONUtils;
|
|
|
+import org.apache.commons.collections4.CollectionUtils;
|
|
|
+import org.apache.commons.collections4.MapUtils;
|
|
|
+import org.apache.commons.lang.exception.ExceptionUtils;
|
|
|
+import org.slf4j.Logger;
|
|
|
+import org.slf4j.LoggerFactory;
|
|
|
+
|
|
|
+import java.util.*;
|
|
|
+import java.util.concurrent.*;
|
|
|
+
|
|
|
+
|
|
|
+public class VlogRovFMScorer extends BaseFMModelScorer {
|
|
|
+
|
|
|
+ private static final int LOCAL_TIME_OUT = 150;
|
|
|
+ private final static Logger LOGGER = LoggerFactory.getLogger(VlogRovFMScorer.class);
|
|
|
+ private static final ExecutorService executorService = Executors.newFixedThreadPool(128);
|
|
|
+
|
|
|
+
|
|
|
+ public VlogRovFMScorer(ScorerConfigInfo configInfo) {
|
|
|
+ super(configInfo);
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public List<RankItem> scoring(final ScoreParam param,
|
|
|
+ final UserFeature userFeature,
|
|
|
+ final List<RankItem> rankItems) {
|
|
|
+ throw new NoSuchMethodError();
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public List<RankItem> scoring(final Map<String, String> sceneFeatureMap,
|
|
|
+ final Map<String, String> userFeatureMap,
|
|
|
+ final List<RankItem> rankItems) {
|
|
|
+ if (CollectionUtils.isEmpty(rankItems)) {
|
|
|
+ return rankItems;
|
|
|
+ }
|
|
|
+
|
|
|
+ long startTime = System.currentTimeMillis();
|
|
|
+ FMModel model = (FMModel) this.getModel();
|
|
|
+ LOGGER.debug("model size: [{}]", model.getModelSize());
|
|
|
+
|
|
|
+ List<RankItem> result = rankItems;
|
|
|
+ result = rankByJava(
|
|
|
+ sceneFeatureMap, userFeatureMap, rankItems
|
|
|
+ );
|
|
|
+
|
|
|
+ LOGGER.debug("ctr ranker time java items size={}, time={} ", result != null ? result.size() : 0,
|
|
|
+ System.currentTimeMillis() - startTime);
|
|
|
+
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+
|
|
|
+ private List<RankItem> rankByJava(final Map<String, String> sceneFeatureMap,
|
|
|
+ final Map<String, String> userFeatureMap,
|
|
|
+ final List<RankItem> items) {
|
|
|
+ long startTime = System.currentTimeMillis();
|
|
|
+ FMModel model = (FMModel) this.getModel();
|
|
|
+ LOGGER.debug("model size: [{}]", model.getModelSize());
|
|
|
+
|
|
|
+ // 所有都参与打分，按照ctr排序
|
|
|
+ multipleCtrScore(items, userFeatureMap, sceneFeatureMap, model);
|
|
|
+
|
|
|
+ // debug log
|
|
|
+ if (LOGGER.isDebugEnabled()) {
|
|
|
+ for (int i = 0; i < items.size(); i++) {
|
|
|
+ LOGGER.debug("before enter feeds model predict ctr score [{}] [{}]", items.get(i), items.get(i));
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ Collections.sort(items);
|
|
|
+
|
|
|
+ LOGGER.debug("ctr ranker java execute time: [{}]", System.currentTimeMillis() - startTime);
|
|
|
+ LOGGER.debug("[ctr ranker time java] items size={}, cost={} ", items != null ? items.size() : 0,
|
|
|
+ System.currentTimeMillis() - startTime);
|
|
|
+ return items;
|
|
|
+ }
|
|
|
+
|
|
|
+ private void multipleCtrScore(final List<RankItem> items,
|
|
|
+ final Map<String, String> userFeatureMap,
|
|
|
+ final Map<String, String> sceneFeatureMap,
|
|
|
+ final FMModel model) {
|
|
|
+
|
|
|
+ List<Callable<Object>> calls = new ArrayList<Callable<Object>>();
|
|
|
+ for (int index = 0; index < items.size(); index++) {
|
|
|
+ final int fIndex = index;
|
|
|
+ calls.add(new Callable<Object>() {
|
|
|
+ @Override
|
|
|
+ public Object call() throws Exception {
|
|
|
+ try {
|
|
|
+ calcScore(model, items.get(fIndex), userFeatureMap, sceneFeatureMap);
|
|
|
+ } catch (Exception e) {
|
|
|
+ LOGGER.error("ctr exception: [{}] [{}]", items.get(fIndex).videoId, ExceptionUtils.getFullStackTrace(e));
|
|
|
+ }
|
|
|
+ return new Object();
|
|
|
+ }
|
|
|
+ });
|
|
|
+ }
|
|
|
+
|
|
|
+ List<Future<Object>> futures = null;
|
|
|
+ try {
|
|
|
+ futures = executorService.invokeAll(calls, LOCAL_TIME_OUT, TimeUnit.MILLISECONDS);
|
|
|
+ } catch (InterruptedException e) {
|
|
|
+ LOGGER.error("execute invoke fail: {}", ExceptionUtils.getFullStackTrace(e));
|
|
|
+ }
|
|
|
+
|
|
|
+ //等待所有请求的结果返回, 超时也返回
|
|
|
+ int cancel = 0;
|
|
|
+ if (futures != null) {
|
|
|
+ for (Future<Object> future : futures) {
|
|
|
+ try {
|
|
|
+ if (!future.isDone() || future.isCancelled() || future.get() == null) {
|
|
|
+ cancel++;
|
|
|
+ }
|
|
|
+ } catch (InterruptedException e) {
|
|
|
+ LOGGER.error("InterruptedException {},{}", ExceptionUtils.getFullStackTrace(e));
|
|
|
+ } catch (ExecutionException e) {
|
|
|
+ LOGGER.error("ExecutionException {},{}", sceneFeatureMap.size(),
|
|
|
+ ExceptionUtils.getFullStackTrace(e));
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ public double calcScore(final FMModel model,
|
|
|
+ final RankItem item,
|
|
|
+ final Map<String, String> userFeatureMap,
|
|
|
+ final Map<String, String> sceneFeatureMap) {
|
|
|
+
|
|
|
+
|
|
|
+ Map<String, String> featureMap = new HashMap<>();
|
|
|
+ if (MapUtils.isNotEmpty(item.getFeatureMap())) {
|
|
|
+ featureMap.putAll(item.getFeatureMap());
|
|
|
+ }
|
|
|
+ if (MapUtils.isNotEmpty(userFeatureMap)) {
|
|
|
+ featureMap.putAll(userFeatureMap);
|
|
|
+ }
|
|
|
+ if (MapUtils.isNotEmpty(sceneFeatureMap)) {
|
|
|
+ featureMap.putAll(sceneFeatureMap);
|
|
|
+ }
|
|
|
+
|
|
|
+ double pro = 0.0;
|
|
|
+ if (MapUtils.isNotEmpty(featureMap)) {
|
|
|
+ try {
|
|
|
+ pro = model.score(featureMap);
|
|
|
+ LOGGER.info("fea : {}, score:{}", JSONUtils.toJson(featureMap), pro);
|
|
|
+ } catch (Exception e) {
|
|
|
+ LOGGER.error("score error for doc={} exception={}", item.getVideoId(), ExceptionUtils.getFullStackTrace(e));
|
|
|
+ }
|
|
|
+ }
|
|
|
+ item.setScoreRov(pro);
|
|
|
+ return pro;
|
|
|
+ }
|
|
|
+}
|
zhangbo