|
|
@@ -0,0 +1,163 @@
|
|
|
+package com.tzld.piaoquan.ad.engine.service.score;
|
|
|
+
|
|
|
+
|
|
|
+import com.tzld.piaoquan.ad.engine.commons.score.AbstractScorer;
|
|
|
+import com.tzld.piaoquan.ad.engine.commons.score.ScoreParam;
|
|
|
+import com.tzld.piaoquan.ad.engine.commons.score.ScorerConfigInfo;
|
|
|
+import com.tzld.piaoquan.ad.engine.commons.score.model.CvrAdjustingModel;
|
|
|
+import com.tzld.piaoquan.recommend.feature.domain.ad.base.AdRankItem;
|
|
|
+import com.tzld.piaoquan.recommend.feature.domain.ad.base.AdRequestContext;
|
|
|
+import com.tzld.piaoquan.recommend.feature.domain.ad.base.UserAdFeature;
|
|
|
+import org.apache.commons.collections4.CollectionUtils;
|
|
|
+import org.apache.commons.lang.exception.ExceptionUtils;
|
|
|
+import org.slf4j.Logger;
|
|
|
+import org.slf4j.LoggerFactory;
|
|
|
+
|
|
|
+import java.util.ArrayList;
|
|
|
+import java.util.List;
|
|
|
+import java.util.Objects;
|
|
|
+import java.util.concurrent.*;
|
|
|
+
|
|
|
+
|
|
|
+//@Service
|
|
|
+public class VlogAdCvrLRAdjustingScorer extends AbstractScorer {
|
|
|
+
|
|
|
+ private static final int LOCAL_TIME_OUT = 150;
|
|
|
+ private final static Logger LOGGER = LoggerFactory.getLogger(VlogAdCvrLRAdjustingScorer.class);
|
|
|
+ private static final ExecutorService executorService = Executors.newFixedThreadPool(8);
|
|
|
+
|
|
|
+ public VlogAdCvrLRAdjustingScorer(ScorerConfigInfo configInfo) {
|
|
|
+ super(configInfo);
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public void loadModel() {
|
|
|
+ doLoadModel(CvrAdjustingModel.class);
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public List<AdRankItem> scoring(final ScoreParam param,
|
|
|
+ final UserAdFeature userFeature,
|
|
|
+ final List<AdRankItem> rankItems) {
|
|
|
+
|
|
|
+ if (CollectionUtils.isEmpty(rankItems)) {
|
|
|
+ return rankItems;
|
|
|
+ }
|
|
|
+
|
|
|
+ long startTime = System.currentTimeMillis();
|
|
|
+ List<AdRankItem> result = rankByJava(rankItems, param.getRequestContext(), userFeature);
|
|
|
+ LOGGER.debug("ctr ranker time java items size={}, time={} ",
|
|
|
+ result.size(), System.currentTimeMillis() - startTime);
|
|
|
+
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+
|
|
|
+ private List<AdRankItem> rankByJava(final List<AdRankItem> items,
|
|
|
+ final AdRequestContext requestContext,
|
|
|
+ final UserAdFeature user) {
|
|
|
+ long startTime = System.currentTimeMillis();
|
|
|
+ CvrAdjustingModel model = (CvrAdjustingModel) this.getModel();
|
|
|
+ LOGGER.debug("model size: [{}]", model.getModelSize());
|
|
|
+
|
|
|
+ // 所有都参与打分，按照cvr排序
|
|
|
+ multipleScore(items, requestContext, model);
|
|
|
+
|
|
|
+ // debug log
|
|
|
+ if (LOGGER.isDebugEnabled()) {
|
|
|
+ for (AdRankItem item : items) {
|
|
|
+ LOGGER.debug("after enter feeds model predict cvr adjusting score [{}] [{}]", item, item.getScore());
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ LOGGER.debug("ctr ranker java execute time: [{}]", System.currentTimeMillis() - startTime);
|
|
|
+ LOGGER.debug("[ctr ranker time java] items size={}, cost={} ",
|
|
|
+ items.size(), System.currentTimeMillis() - startTime);
|
|
|
+ return items;
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 校准cvr
|
|
|
+ */
|
|
|
+ public void calcScore(final CvrAdjustingModel lrModel,
|
|
|
+ final AdRankItem item,
|
|
|
+ final AdRequestContext requestContext) {
|
|
|
+
|
|
|
+ double pro = item.getCvr();
|
|
|
+ try {
|
|
|
+ Double coef = lrModel.getAdjustingCoefficien(pro);
|
|
|
+ if (Objects.nonNull(coef)) {
|
|
|
+ LOGGER.info("[VlogAdCvrLRAdjustingScorer.cvr adjusting] before: {}", pro);
|
|
|
+ pro = pro / coef;
|
|
|
+ LOGGER.info("[VlogAdCvrLRAdjustingScorer.cvr adjusting] after: {}, coef: {}", pro, coef);
|
|
|
+
|
|
|
+ }
|
|
|
+
|
|
|
+ } catch (
|
|
|
+ Exception e) {
|
|
|
+ LOGGER.error("score error for doc={} exception={}",
|
|
|
+ item.getAdId(), ExceptionUtils.getFullStackTrace(e));
|
|
|
+ }
|
|
|
+ item.setCvr(pro);
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 并行打分
|
|
|
+ *
|
|
|
+ * @param items
|
|
|
+ * @param userInfoBytes
|
|
|
+ * @param requestContext
|
|
|
+ * @param model
|
|
|
+ */
|
|
|
+ private void multipleScore(final List<AdRankItem> items,
|
|
|
+ final AdRequestContext requestContext,
|
|
|
+ final CvrAdjustingModel 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), requestContext);
|
|
|
+ } catch (
|
|
|
+ Exception e) {
|
|
|
+ LOGGER.error("ctr exception: [{}] [{}]", items.get(fIndex).adId, 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 {},{}",
|
|
|
+ requestContext.getApptype(), ExceptionUtils.getFullStackTrace(e));
|
|
|
+ } catch (
|
|
|
+ ExecutionException e) {
|
|
|
+ LOGGER.error("ExecutionException {},{}",
|
|
|
+ requestContext.getApptype(), ExceptionUtils.getFullStackTrace(e));
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ LOGGER.debug("Ctr Score {}, Total: {}, Cancel: {}", requestContext.getApptype(), items.size(), cancel);
|
|
|
+ }
|
|
|
+}
|
