|
|
@@ -0,0 +1,331 @@
|
|
|
+package com.tzld.piaoquan.recommend.server.implement.score;
|
|
|
+
|
|
|
+
|
|
|
+import com.tzld.piaoquan.recommend.feature.domain.video.base.*;
|
|
|
+import com.tzld.piaoquan.recommend.feature.domain.video.feature.VlogShareLRFeatureExtractor;
|
|
|
+import com.tzld.piaoquan.recommend.feature.model.sample.LRSamples;
|
|
|
+import com.tzld.piaoquan.recommend.server.common.base.RankItem;
|
|
|
+import com.tzld.piaoquan.recommend.server.service.rank.strategy.OfflineVlogShareLRFeatureExtractor;
|
|
|
+import com.tzld.piaoquan.recommend.server.service.score.BaseLRModelScorer;
|
|
|
+import com.tzld.piaoquan.recommend.server.service.score.ScoreParam;
|
|
|
+import com.tzld.piaoquan.recommend.server.service.score.ScorerConfigInfo;
|
|
|
+import com.tzld.piaoquan.recommend.server.service.score.model.LRModel;
|
|
|
+import org.apache.commons.collections4.CollectionUtils;
|
|
|
+import org.apache.commons.lang.exception.ExceptionUtils;
|
|
|
+import org.slf4j.Logger;
|
|
|
+import org.slf4j.LoggerFactory;
|
|
|
+
|
|
|
+import java.util.*;
|
|
|
+import java.util.concurrent.*;
|
|
|
+
|
|
|
+
|
|
|
+public class VlogShareLRScorer4Ros extends BaseLRModelScorer {
|
|
|
+
|
|
|
+ private final static int CORE_POOL_SIZE = 64;
|
|
|
+
|
|
|
+ private static final int LOCAL_TIME_OUT = 150;
|
|
|
+ private final static Logger LOGGER = LoggerFactory.getLogger(VlogShareLRScorer4Ros.class);
|
|
|
+ private static final ExecutorService executorService = Executors.newFixedThreadPool(128);
|
|
|
+ private static final double defaultUserCtrGroupNumber = 10.0;
|
|
|
+ private static final int enterFeedsScoreRatio = 10;
|
|
|
+ private static final int enterFeedsScoreNum = 20;
|
|
|
+
|
|
|
+
|
|
|
+ public VlogShareLRScorer4Ros(ScorerConfigInfo configInfo) {
|
|
|
+ super(configInfo);
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public List<RankItem> scoring(final ScoreParam param,
|
|
|
+ final UserFeature userFeature,
|
|
|
+ final List<RankItem> rankItems) {
|
|
|
+
|
|
|
+ if (CollectionUtils.isEmpty(rankItems)) {
|
|
|
+ return rankItems;
|
|
|
+ }
|
|
|
+
|
|
|
+ long startTime = System.currentTimeMillis();
|
|
|
+ LRModel model = (LRModel) this.getModel();
|
|
|
+ LOGGER.debug("model size: [{}]", model.getModelSize());
|
|
|
+
|
|
|
+ List<RankItem> result = rankItems;
|
|
|
+ result = rankByJava(rankItems, param.getRequestContext(),
|
|
|
+ userFeature == null ? UserFeature.defaultInstance(param.getMid()) : userFeature);
|
|
|
+
|
|
|
+ LOGGER.debug("ctr ranker time java items size={}, time={} ", result != null ? result.size() : 0,
|
|
|
+ System.currentTimeMillis() - startTime);
|
|
|
+
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+
|
|
|
+ private List<RankItem> rankByJava(final List<RankItem> items,
|
|
|
+ final RequestContext requestContext,
|
|
|
+ final UserFeature user) {
|
|
|
+ long startTime = System.currentTimeMillis();
|
|
|
+ LRModel model = (LRModel) this.getModel();
|
|
|
+ LOGGER.debug("model size: [{}]", model.getModelSize());
|
|
|
+
|
|
|
+ // userBytes
|
|
|
+ UserBytesFeature userInfoBytes = null;
|
|
|
+ userInfoBytes = new UserBytesFeature(user);
|
|
|
+
|
|
|
+ // 所有都参与打分，按照ctr排序
|
|
|
+ multipleCtrScore(items, userInfoBytes, requestContext, 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;
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 计算 predict ctr
|
|
|
+ */
|
|
|
+ public double calcScore(final LRModel lrModel,
|
|
|
+ final RankItem item,
|
|
|
+ final UserBytesFeature userInfoBytes,
|
|
|
+ final RequestContext requestContext) {
|
|
|
+
|
|
|
+ LRSamples lrSamples = null;
|
|
|
+ VlogShareLRFeatureExtractor bytesFeatureExtractor;
|
|
|
+ bytesFeatureExtractor = new VlogShareLRFeatureExtractor();
|
|
|
+
|
|
|
+ try {
|
|
|
+ VideoBytesFeature newsInfoBytes = new VideoBytesFeature(item.getItemFeature() == null
|
|
|
+ ? ItemFeature.defaultInstance(item.getVideoId() + "")
|
|
|
+ : item.getItemFeature());
|
|
|
+ lrSamples = bytesFeatureExtractor.single(userInfoBytes, newsInfoBytes,
|
|
|
+ new RequestContextBytesFeature(requestContext));
|
|
|
+ } catch (Exception e) {
|
|
|
+ LOGGER.error("extract feature error for imei={}, doc={}, [{}]", new Object[]{new String(userInfoBytes.getUid()), item.getVideoId(),
|
|
|
+ ExceptionUtils.getFullStackTrace(e)});
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+ double pro = 0.0;
|
|
|
+ if (lrSamples != null && lrSamples.getFeaturesList() != null) {
|
|
|
+ try {
|
|
|
+ pro = lrModel.score(lrSamples);
|
|
|
+ } catch (Exception e) {
|
|
|
+ LOGGER.error("score error for doc={} exception={}", new Object[]{
|
|
|
+ item.getVideoId(), ExceptionUtils.getFullStackTrace(e)});
|
|
|
+ }
|
|
|
+ // 增加实时特征后打开在线存储日志逻辑
|
|
|
+ //
|
|
|
+ // CtrSamples.Builder samples = com.tzld.piaoquan.recommend.server.gen.recommend.CtrSamples.newBuilder();
|
|
|
+ // samples.setLr_samples(lrSamples);
|
|
|
+ // item.setSamples(samples);
|
|
|
+ //
|
|
|
+ }
|
|
|
+ item.setScore(pro);
|
|
|
+ return pro;
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 并行打分
|
|
|
+ *
|
|
|
+ * @param items
|
|
|
+ * @param userInfoBytes
|
|
|
+ * @param requestContext
|
|
|
+ * @param model
|
|
|
+ */
|
|
|
+ private void multipleCtrScore(final List<RankItem> items,
|
|
|
+ final UserBytesFeature userInfoBytes,
|
|
|
+ final RequestContext requestContext,
|
|
|
+ final LRModel model) {
|
|
|
+
|
|
|
+ List<Callable<Object>> calls = new ArrayList<Callable<Object>>();
|
|
|
+ for (int index = 0; index < items.size(); index++) {
|
|
|
+ final int fIndex = index;
|
|
|
+ items.get(fIndex).setScore(0.0); //原始分为 cube中的粗打分，如果超时，为原始值存在问题， 需要置0
|
|
|
+ calls.add(new Callable<Object>() {
|
|
|
+ @Override
|
|
|
+ public Object call() throws Exception {
|
|
|
+ try {
|
|
|
+ calcScore(model, items.get(fIndex), userInfoBytes, requestContext);
|
|
|
+ } 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 {},{}", requestContext.getRequest_id(),
|
|
|
+ ExceptionUtils.getFullStackTrace(e));
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ LOGGER.debug("Ctr Score {}, Total: {}, Cancel: {}", new Object[]{requestContext.getRequest_id(), items.size(), cancel});
|
|
|
+ }
|
|
|
+ @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();
|
|
|
+ LRModel model = (LRModel) 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();
|
|
|
+ LRModel model = (LRModel) this.getModel();
|
|
|
+ LOGGER.debug("model size: [{}]", model.getModelSize());
|
|
|
+ // userBytes
|
|
|
+ Map<String, byte[]> userFeatureMapByte = new HashMap<>();
|
|
|
+ for(Map.Entry<String, String> entry: userFeatureMap.entrySet()){
|
|
|
+ userFeatureMapByte.put(entry.getKey(), entry.getValue().getBytes());
|
|
|
+ }
|
|
|
+ //sceneBytes
|
|
|
+ Map<String, byte[]> sceneFeatureMapByte = new HashMap<>();
|
|
|
+ for(Map.Entry<String, String> entry: sceneFeatureMap.entrySet()){
|
|
|
+ sceneFeatureMapByte.put(entry.getKey(), entry.getValue().getBytes());
|
|
|
+ }
|
|
|
+
|
|
|
+ // 所有都参与打分，按照ctr排序
|
|
|
+ multipleCtrScore(items, userFeatureMapByte, sceneFeatureMapByte, 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, byte[]> userFeatureMapByte,
|
|
|
+ final Map<String, byte[]> sceneFeatureMapByte,
|
|
|
+ final LRModel model) {
|
|
|
+
|
|
|
+ List<Callable<Object>> calls = new ArrayList<Callable<Object>>();
|
|
|
+ for (int index = 0; index < items.size(); index++) {
|
|
|
+ final int fIndex = index;
|
|
|
+// items.get(fIndex).setScore(0.0); //原始分为 cube中的粗打分，如果超时，为原始值存在问题， 需要置0
|
|
|
+ calls.add(new Callable<Object>() {
|
|
|
+ @Override
|
|
|
+ public Object call() throws Exception {
|
|
|
+ try {
|
|
|
+ calcScore(model, items.get(fIndex), userFeatureMapByte, sceneFeatureMapByte);
|
|
|
+ } 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 {},{}", sceneFeatureMapByte.size(),
|
|
|
+ ExceptionUtils.getFullStackTrace(e));
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ LOGGER.debug("Ctr Score {}, Total: {}, Cancel: {}", new Object[]{sceneFeatureMapByte.size(), items.size(), cancel});
|
|
|
+ }
|
|
|
+
|
|
|
+ public double calcScore(final LRModel lrModel,
|
|
|
+ final RankItem item,
|
|
|
+ final Map<String, byte[]> userFeatureMapByte,
|
|
|
+ final Map<String, byte[]> sceneFeatureMapByte) {
|
|
|
+
|
|
|
+ LRSamples lrSamples = null;
|
|
|
+ OfflineVlogShareLRFeatureExtractor bytesFeatureExtractor;
|
|
|
+ bytesFeatureExtractor = new OfflineVlogShareLRFeatureExtractor();
|
|
|
+
|
|
|
+ try {
|
|
|
+
|
|
|
+ Map<String, byte[]> itemFeatureByte = new HashMap<>();
|
|
|
+ for (Map.Entry<String, String> entry: item.getFeatureMap().entrySet()){
|
|
|
+ itemFeatureByte.put(entry.getKey(), entry.getValue().getBytes());
|
|
|
+ }
|
|
|
+ lrSamples = bytesFeatureExtractor.single(userFeatureMapByte, itemFeatureByte, sceneFeatureMapByte);
|
|
|
+ } catch (Exception e) {
|
|
|
+ LOGGER.error("extract feature error for imei={}, doc={}, [{}]", new Object[]{"", item.getVideoId(),
|
|
|
+ ExceptionUtils.getFullStackTrace(e)});
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+ double pro = 0.0;
|
|
|
+ if (lrSamples != null && lrSamples.getFeaturesList() != null) {
|
|
|
+ try {
|
|
|
+ pro = lrModel.score(lrSamples);
|
|
|
+ } catch (Exception e) {
|
|
|
+ LOGGER.error("score error for doc={} exception={}", new Object[]{
|
|
|
+ item.getVideoId(), ExceptionUtils.getFullStackTrace(e)});
|
|
|
+ }
|
|
|
+ }
|
|
|
+ item.setScoreRos(pro);
|
|
|
+ return pro;
|
|
|
+ }
|
|
|
+}
|
sunmingze