|
|
@@ -0,0 +1,735 @@
|
|
|
+package com.aliyun.odps.spark.examples.makedata_ad.v20240718
|
|
|
+
|
|
|
+import com.alibaba.fastjson.{JSON, JSONObject}
|
|
|
+import com.aliyun.odps.TableSchema
|
|
|
+import com.aliyun.odps.data.Record
|
|
|
+import com.aliyun.odps.spark.examples.myUtils.{MyDateUtils, ParamUtils, env}
|
|
|
+import examples.extractor.{ExtractorUtils, RankExtractorFeature_20240530}
|
|
|
+import examples.utils.{AdUtil, DateTimeUtil, SimilarityUtils}
|
|
|
+import org.apache.spark.sql.SparkSession
|
|
|
+
|
|
|
+import java.time.{Instant, ZoneId, ZonedDateTime}
|
|
|
+import scala.collection.JavaConversions._
|
|
|
+import scala.collection.mutable.ArrayBuffer
|
|
|
+import scala.io.Source
|
|
|
+import scala.language.postfixOps
|
|
|
+import scala.util.Random
|
|
|
+
|
|
|
+object makedata_ad_33_bucketDataFromOriginToHive_20250522 {
|
|
|
+ val CTR_SMOOTH_BETA_FACTOR = 25
|
|
|
+ val CVR_SMOOTH_BETA_FACTOR = 10
|
|
|
+ val CTCVR_SMOOTH_BETA_FACTOR = 100
|
|
|
+
|
|
|
+ def main(args: Array[String]): Unit = {
|
|
|
+ val spark = SparkSession
|
|
|
+ .builder()
|
|
|
+ .appName(this.getClass.getName)
|
|
|
+ .getOrCreate()
|
|
|
+ val sc = spark.sparkContext
|
|
|
+
|
|
|
+
|
|
|
+ // 1 读取参数
|
|
|
+ val param = ParamUtils.parseArgs(args)
|
|
|
+ val tablePart = param.getOrElse("tablePart", "64").toInt
|
|
|
+ val beginStr = param.getOrElse("beginStr", "20250216")
|
|
|
+ val endStr = param.getOrElse("endStr", "20250216")
|
|
|
+ val project = param.getOrElse("project", "loghubods")
|
|
|
+ val inputTable = param.getOrElse("inputTable", "alg_recsys_ad_sample_all")
|
|
|
+ val outputTable = param.getOrElse("outputTable", "ad_easyrec_train_data_v1_sampled")
|
|
|
+ val outputTable2 = param.getOrElse("outputTable2", "")
|
|
|
+ val filterHours = param.getOrElse("filterHours", "00,01,02,03,04,05,06,07").split(",").toSet
|
|
|
+ val idDefaultValue = param.getOrElse("idDefaultValue", "1.0").toDouble
|
|
|
+ val filterNames = param.getOrElse("filterNames", "").split(",").filter(_.nonEmpty).toSet
|
|
|
+ val filterAdverIds = param.getOrElse("filterAdverIds", "").split(",").filter(_.nonEmpty).toSet
|
|
|
+ val whatLabel = param.getOrElse("whatLabel", "ad_is_conversion")
|
|
|
+ val negSampleRate = param.getOrElse("negSampleRate", "1").toDouble
|
|
|
+ // 分割样本集的比例，splitRate部分输出至outputTable，补集输出至outputTable2（如果outputTable2不为空）
|
|
|
+ val splitRate = param.getOrElse("splitRate", "0.9").toDouble
|
|
|
+
|
|
|
+ val loader = getClass.getClassLoader
|
|
|
+ val resourceUrlBucket = loader.getResource("20250217_ad_bucket_688.txt")
|
|
|
+ val buckets =
|
|
|
+ if (resourceUrlBucket != null) {
|
|
|
+ val buckets = Source.fromURL(resourceUrlBucket).getLines().mkString("\n")
|
|
|
+ Source.fromURL(resourceUrlBucket).close()
|
|
|
+ buckets
|
|
|
+ } else {
|
|
|
+ ""
|
|
|
+ }
|
|
|
+ val bucketsMap = buckets.split("\n")
|
|
|
+ .map(r => r.replace(" ", "").replaceAll("\n", ""))
|
|
|
+ .filter(r => r.nonEmpty)
|
|
|
+ .map(r => {
|
|
|
+ val rList = r.split("\t")
|
|
|
+ val featureName = rList(0).replace("*", "_x_").replace("(view)", "_view")
|
|
|
+ (featureName, (rList(1).toDouble, rList(2).split(",").map(_.toDouble)))
|
|
|
+ }).toMap
|
|
|
+ val bucketsMap_br = sc.broadcast(bucketsMap)
|
|
|
+ val denseFeatureNames = bucketsMap.keySet
|
|
|
+ val sparseFeatureNames = Set(
|
|
|
+ "cid", "adid", "adverid", "targeting_conversion",
|
|
|
+ "region", "city", "brand",
|
|
|
+ "vid", "cate1", "cate2",
|
|
|
+ "user_cid_click_list", "user_cid_conver_list",
|
|
|
+ "user_vid_return_tags_2h", "user_vid_return_tags_1d", "user_vid_return_tags_3d", "user_vid_return_tags_7d",
|
|
|
+ "user_vid_return_tags_14d", "apptype", "ts", "mid", "pqtid", "hour", "hour_quarter", "root_source_scene",
|
|
|
+ "root_source_channel", "is_first_layer", "title_split", "profession", "user_vid_share_tags_1d", "user_vid_share_tags_14d",
|
|
|
+ "user_vid_return_cate1_14d", "user_vid_return_cate2_14d", "user_vid_share_cate1_14d", "user_vid_share_cate2_14d",
|
|
|
+ "creative_type", "creative_hook_embedding", "creative_why_embedding", "creative_action_embedding", "user_has_conver_1y",
|
|
|
+ "user_adverid_view_3d", "user_adverid_view_7d", "user_adverid_view_30d",
|
|
|
+ "user_adverid_click_3d", "user_adverid_click_7d", "user_adverid_click_30d",
|
|
|
+ "user_adverid_conver_3d", "user_adverid_conver_7d", "user_adverid_conver_30d",
|
|
|
+ "user_skuid_view_3d", "user_skuid_view_7d", "user_skuid_view_30d",
|
|
|
+ "user_skuid_click_3d", "user_skuid_click_7d", "user_skuid_click_30d",
|
|
|
+ "user_skuid_conver_3d", "user_skuid_conver_7d", "user_skuid_conver_30d",
|
|
|
+ "is_weekday", "day_of_the_week")
|
|
|
+
|
|
|
+
|
|
|
+ // 2 读取odps+表信息
|
|
|
+ val odpsOps = env.getODPS(sc)
|
|
|
+ // 3 循环执行数据生产
|
|
|
+ val dateRange = MyDateUtils.getDateRange(beginStr, endStr)
|
|
|
+ for (dt <- dateRange) {
|
|
|
+ val timeRange = MyDateUtils.getDateHourRange(dt + "08", dt + "23")
|
|
|
+ val recordRdd = timeRange.map { dt_hh =>
|
|
|
+ val dt = dt_hh.substring(0, 8)
|
|
|
+ val hh = dt_hh.substring(8, 10)
|
|
|
+ val partition = s"dt=$dt,hh=$hh"
|
|
|
+ if (filterHours.nonEmpty && filterHours.contains(hh)) {
|
|
|
+ None
|
|
|
+ } else {
|
|
|
+ Some(partition)
|
|
|
+ }
|
|
|
+ }.collect {
|
|
|
+ case Some(partition) => partition
|
|
|
+ }.map(partition => {
|
|
|
+ val odpsData = odpsOps.readTable(project = project,
|
|
|
+ table = inputTable,
|
|
|
+ partition = partition,
|
|
|
+ transfer = func,
|
|
|
+ numPartition = tablePart)
|
|
|
+ .filter(record => {
|
|
|
+ AdUtil.isApi(record)
|
|
|
+ })
|
|
|
+ .filter(record => {
|
|
|
+ val appType = record.getString("apptype")
|
|
|
+ !Set("12", "13").contains(appType)
|
|
|
+ })
|
|
|
+ .filter(record => {
|
|
|
+ val adverId = record.getString("adverid")
|
|
|
+ !filterAdverIds.contains(adverId)
|
|
|
+ })
|
|
|
+ .filter(record => {
|
|
|
+ val label = record.getString(whatLabel).toInt
|
|
|
+ label > 0 || Random.nextDouble() < negSampleRate
|
|
|
+ })
|
|
|
+ .map(record => {
|
|
|
+ val featureMap = new JSONObject()
|
|
|
+ val ts = record.getString("ts").toInt
|
|
|
+ val instant = Instant.ofEpochSecond(ts)
|
|
|
+ // 设置时区为中国时区
|
|
|
+ val chinaZone = ZoneId.of("Asia/Shanghai")
|
|
|
+ // 将 Instant 对象转换为中国时区的 ZonedDateTime 对象
|
|
|
+ val zonedDateTime = ZonedDateTime.ofInstant(instant, chinaZone)
|
|
|
+ // 获取星期几（1=周一，7=周日）
|
|
|
+ val dayOfTheWeek = zonedDateTime.getDayOfWeek.getValue()
|
|
|
+ val isWeekday = if (dayOfTheWeek <= 5) 1 else 2
|
|
|
+ val cid = record.getString("cid")
|
|
|
+ val mid = record.getString("mid")
|
|
|
+ val pqtid = record.getString("pqtid")
|
|
|
+ val apptype = record.getString("apptype")
|
|
|
+
|
|
|
+ featureMap.put("apptype", apptype)
|
|
|
+ featureMap.put("ts", ts)
|
|
|
+ featureMap.put("mid", mid)
|
|
|
+ featureMap.put("pqtid", pqtid)
|
|
|
+ val extend: JSONObject = if (record.isNull("extend")) new JSONObject() else
|
|
|
+ JSON.parseObject(record.getString("extend"))
|
|
|
+ val mateFeature: JSONObject = if (record.isNull("metafeaturemap")) new JSONObject() else
|
|
|
+ JSON.parseObject(record.getString("metafeaturemap"))
|
|
|
+ val reqFeature: JSONObject = if (!mateFeature.containsKey("reqFeature")) new JSONObject() else
|
|
|
+ mateFeature.getJSONObject("reqFeature")
|
|
|
+ val sceneFeature: JSONObject = if (!mateFeature.containsKey("sceneFeature")) new JSONObject() else
|
|
|
+ mateFeature.getJSONObject("sceneFeature")
|
|
|
+ val b1: JSONObject = if (!mateFeature.containsKey("alg_cid_feature_basic_info")) new JSONObject() else
|
|
|
+ mateFeature.getJSONObject("alg_cid_feature_basic_info")
|
|
|
+ val b2: JSONObject = if (!mateFeature.containsKey("alg_cid_feature_adver_action")) new JSONObject() else
|
|
|
+ mateFeature.getJSONObject("alg_cid_feature_adver_action")
|
|
|
+ val b3: JSONObject = if (!mateFeature.containsKey("alg_cid_feature_cid_action")) new JSONObject() else
|
|
|
+ mateFeature.getJSONObject("alg_cid_feature_cid_action")
|
|
|
+ val b4: JSONObject = if (!mateFeature.containsKey("alg_cid_feature_region_action")) new JSONObject() else
|
|
|
+ mateFeature.getJSONObject("alg_cid_feature_region_action")
|
|
|
+ val b5: JSONObject = if (!mateFeature.containsKey("alg_cid_feature_app_action")) new JSONObject() else
|
|
|
+ mateFeature.getJSONObject("alg_cid_feature_app_action")
|
|
|
+ val b6: JSONObject = if (!mateFeature.containsKey("alg_cid_feature_week_action")) new JSONObject() else
|
|
|
+ mateFeature.getJSONObject("alg_cid_feature_week_action")
|
|
|
+ val b7: JSONObject = if (!mateFeature.containsKey("alg_cid_feature_hour_action")) new JSONObject() else
|
|
|
+ mateFeature.getJSONObject("alg_cid_feature_hour_action")
|
|
|
+ val b8: JSONObject = if (!mateFeature.containsKey("alg_cid_feature_brand_action")) new JSONObject() else
|
|
|
+ mateFeature.getJSONObject("alg_cid_feature_brand_action")
|
|
|
+ val b9: JSONObject = if (!mateFeature.containsKey("alg_cid_feature_weChatVersion_action")) new JSONObject() else
|
|
|
+ mateFeature.getJSONObject("alg_cid_feature_weChatVersion_action")
|
|
|
+
|
|
|
+ featureMap.put("cid_" + cid, idDefaultValue)
|
|
|
+ if (b1.containsKey("adid") && b1.getString("adid").nonEmpty) {
|
|
|
+ featureMap.put("adid_" + b1.getString("adid"), idDefaultValue)
|
|
|
+ }
|
|
|
+ if (b1.containsKey("adverid") && b1.getString("adverid").nonEmpty) {
|
|
|
+ featureMap.put("adverid_" + b1.getString("adverid"), idDefaultValue)
|
|
|
+ }
|
|
|
+ if (b1.containsKey("targeting_conversion") && b1.getString("targeting_conversion").nonEmpty) {
|
|
|
+ featureMap.put("targeting_conversion_" + b1.getString("targeting_conversion"), idDefaultValue)
|
|
|
+ }
|
|
|
+ if (b1.containsKey("cid") && b1.getString("cid").nonEmpty) {
|
|
|
+ featureMap.put("cid", b1.getBigInteger("cid"))
|
|
|
+ }
|
|
|
+ if (b1.containsKey("adid") && b1.getString("adid").nonEmpty) {
|
|
|
+ featureMap.put("adid", b1.getBigInteger("adid"))
|
|
|
+ }
|
|
|
+ if (b1.containsKey("adverid") && b1.getString("adverid").nonEmpty) {
|
|
|
+ featureMap.put("adverid", b1.getBigInteger("adverid"))
|
|
|
+ }
|
|
|
+ if (b1.containsKey("profession") && b1.getString("profession").nonEmpty) {
|
|
|
+ featureMap.put("profession", b1.getString("profession"))
|
|
|
+ }
|
|
|
+ if (b1.containsKey("creative_type") && b1.getString("creative_type").nonEmpty) {
|
|
|
+ featureMap.put("creative_type", b1.getString("creative_type"))
|
|
|
+ }
|
|
|
+ if (b1.containsKey("creative_hook_embedding") && b1.getString("creative_hook_embedding").nonEmpty) {
|
|
|
+ featureMap.put("creative_hook_embedding", b1.getString("creative_hook_embedding").split('|').map(_.toDouble).map(_.toFloat).mkString("|"))
|
|
|
+ }
|
|
|
+ if (b1.containsKey("creative_why_embedding") && b1.getString("creative_why_embedding").nonEmpty) {
|
|
|
+ featureMap.put("creative_why_embedding", b1.getString("creative_why_embedding").split('|').map(_.toDouble).map(_.toFloat).mkString("|"))
|
|
|
+ }
|
|
|
+ if (b1.containsKey("creative_action_embedding") && b1.getString("creative_action_embedding").nonEmpty) {
|
|
|
+ featureMap.put("creative_action_embedding", b1.getString("creative_action_embedding").split('|').map(_.toDouble).map(_.toFloat).mkString("|"))
|
|
|
+ }
|
|
|
+
|
|
|
+ if (sceneFeature.containsKey("hour") && sceneFeature.getString("hour").nonEmpty) {
|
|
|
+ featureMap.put("hour", sceneFeature.getString("hour"))
|
|
|
+ }
|
|
|
+ if (sceneFeature.containsKey("hour_quarter") && sceneFeature.getString("hour_quarter").nonEmpty) {
|
|
|
+ featureMap.put("hour_quarter", sceneFeature.getString("hour_quarter"))
|
|
|
+ }
|
|
|
+ featureMap.put("is_weekday", isWeekday)
|
|
|
+ featureMap.put("day_of_the_week", dayOfTheWeek)
|
|
|
+
|
|
|
+ val hour = DateTimeUtil.getHourByTimestamp(ts)
|
|
|
+ featureMap.put("hour_" + hour, idDefaultValue)
|
|
|
+
|
|
|
+ val dayOfWeek = DateTimeUtil.getDayOrWeekByTimestamp(ts)
|
|
|
+ featureMap.put("dayofweek_" + dayOfWeek, idDefaultValue);
|
|
|
+
|
|
|
+ featureMap.put("apptype_" + apptype, idDefaultValue);
|
|
|
+
|
|
|
+ if (extend.containsKey("abcode") && extend.getString("abcode").nonEmpty) {
|
|
|
+ featureMap.put("abcode_" + extend.getString("abcode"), idDefaultValue)
|
|
|
+ }
|
|
|
+
|
|
|
+ if (reqFeature.containsKey("region") && reqFeature.getString("region").nonEmpty) {
|
|
|
+ featureMap.put("region", reqFeature.getString("region"))
|
|
|
+ }
|
|
|
+
|
|
|
+ if (reqFeature.containsKey("city") && reqFeature.getString("city").nonEmpty) {
|
|
|
+ featureMap.put("city", reqFeature.getString("city"))
|
|
|
+ }
|
|
|
+
|
|
|
+ if (reqFeature.containsKey("is_first_layer") && reqFeature.getString("is_first_layer").nonEmpty) {
|
|
|
+ featureMap.put("is_first_layer", reqFeature.getString("is_first_layer"))
|
|
|
+ }
|
|
|
+
|
|
|
+ if (reqFeature.containsKey("root_source_scene") && reqFeature.getString("root_source_scene").nonEmpty) {
|
|
|
+ featureMap.put("root_source_scene", reqFeature.getString("root_source_scene"))
|
|
|
+ }
|
|
|
+
|
|
|
+ if (reqFeature.containsKey("root_source_channel") && reqFeature.getString("root_source_channel").nonEmpty) {
|
|
|
+ featureMap.put("root_source_channel", reqFeature.getString("root_source_channel"))
|
|
|
+ }
|
|
|
+
|
|
|
+ if (reqFeature.containsKey("brand") && reqFeature.getString("brand").nonEmpty) {
|
|
|
+ featureMap.put("brand", reqFeature.getString("brand"))
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+ if (b1.containsKey("cpa")) {
|
|
|
+ featureMap.put("cpa", b1.getString("cpa").toDouble)
|
|
|
+ }
|
|
|
+ if (b1.containsKey("weight") && b1.getString("weight").nonEmpty) {
|
|
|
+ featureMap.put("weight", b1.getString("weight").toDouble)
|
|
|
+ }
|
|
|
+
|
|
|
+ for ((bn, prefix1) <- List(
|
|
|
+ (b2, "b2"), (b3, "b3"), (b4, "b4"), (b5, "b5"), (b8, "b8"), (b9, "b9")
|
|
|
+ )) {
|
|
|
+ for (prefix2 <- List(
|
|
|
+ "3h", "6h", "12h", "1d", "3d", "7d", "today", "yesterday"
|
|
|
+ )) {
|
|
|
+ val view = if (bn.isEmpty) 0D else bn.getIntValue("ad_view_" + prefix2).toDouble
|
|
|
+ val click = if (bn.isEmpty) 0D else bn.getIntValue("ad_click_" + prefix2).toDouble
|
|
|
+ val conver = if (bn.isEmpty) 0D else bn.getIntValue("ad_conversion_" + prefix2).toDouble
|
|
|
+ val income = if (bn.isEmpty) 0D else bn.getIntValue("ad_income_" + prefix2).toDouble
|
|
|
+ // NOTE(zhoutian):
|
|
|
+ // 这里cpc只是为了计算cpm的平滑的工具量，没有实际业务意义，因为cpm并非比率，本身不适合直接计算Wilson平滑
|
|
|
+ // 不使用cpa的原因是未来可能出现广告采用cpc计费的情况或者无法获取转化量的情况，用点击更为稳定
|
|
|
+ // 其它几组特征亦采用相同逻辑
|
|
|
+ // 2025-02-17改为增加固定分母平滑，income实际已经可以直接参与cpm平滑计算
|
|
|
+ val cpc = if (click == 0) 0D else income / click
|
|
|
+ val f1 = RankExtractorFeature_20240530.divSmooth2(click, view, CTR_SMOOTH_BETA_FACTOR)
|
|
|
+ val f2 = RankExtractorFeature_20240530.divSmooth2(conver, view, CTCVR_SMOOTH_BETA_FACTOR)
|
|
|
+ val f3 = RankExtractorFeature_20240530.divSmooth2(conver, click, CVR_SMOOTH_BETA_FACTOR)
|
|
|
+ val f4 = conver
|
|
|
+ val f5 = RankExtractorFeature_20240530.divSmooth2(click, view, CTR_SMOOTH_BETA_FACTOR) * cpc * 1000
|
|
|
+ featureMap.put(prefix1 + "_" + prefix2 + "_" + "ctr", f1)
|
|
|
+ featureMap.put(prefix1 + "_" + prefix2 + "_" + "ctcvr", f2)
|
|
|
+ featureMap.put(prefix1 + "_" + prefix2 + "_" + "cvr", f3)
|
|
|
+ featureMap.put(prefix1 + "_" + prefix2 + "_" + "conver", f4)
|
|
|
+ featureMap.put(prefix1 + "_" + prefix2 + "_" + "ecpm", f5)
|
|
|
+
|
|
|
+ featureMap.put(prefix1 + "_" + prefix2 + "_" + "click", click)
|
|
|
+ featureMap.put(prefix1 + "_" + prefix2 + "_" + "conver_x_log_view", conver * RankExtractorFeature_20240530.calLog(view))
|
|
|
+ featureMap.put(prefix1 + "_" + prefix2 + "_" + "conver_x_ctcvr", conver * f2)
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ for ((bn, prefix1) <- List(
|
|
|
+ (b6, "b6"), (b7, "b7")
|
|
|
+ )) {
|
|
|
+ for (prefix2 <- List(
|
|
|
+ "7d", "14d"
|
|
|
+ )) {
|
|
|
+ val view = if (bn.isEmpty) 0D else bn.getIntValue("ad_view_" + prefix2).toDouble
|
|
|
+ val click = if (bn.isEmpty) 0D else bn.getIntValue("ad_click_" + prefix2).toDouble
|
|
|
+ val conver = if (bn.isEmpty) 0D else bn.getIntValue("ad_conversion_" + prefix2).toDouble
|
|
|
+ val income = if (bn.isEmpty) 0D else bn.getIntValue("ad_income_" + prefix2).toDouble
|
|
|
+ val cpc = if (click == 0) 0D else income / click
|
|
|
+ val f1 = RankExtractorFeature_20240530.divSmooth2(click, view, CTR_SMOOTH_BETA_FACTOR)
|
|
|
+ val f2 = RankExtractorFeature_20240530.divSmooth2(conver, view, CTCVR_SMOOTH_BETA_FACTOR)
|
|
|
+ val f3 = RankExtractorFeature_20240530.divSmooth2(conver, click, CVR_SMOOTH_BETA_FACTOR)
|
|
|
+ val f4 = conver
|
|
|
+ val f5 = RankExtractorFeature_20240530.divSmooth2(click, view, CTR_SMOOTH_BETA_FACTOR) * cpc * 1000
|
|
|
+ featureMap.put(prefix1 + "_" + prefix2 + "_" + "ctr", f1)
|
|
|
+ featureMap.put(prefix1 + "_" + prefix2 + "_" + "ctcvr", f2)
|
|
|
+ featureMap.put(prefix1 + "_" + prefix2 + "_" + "cvr", f3)
|
|
|
+ featureMap.put(prefix1 + "_" + prefix2 + "_" + "conver", f4)
|
|
|
+ featureMap.put(prefix1 + "_" + prefix2 + "_" + "ecpm", f5)
|
|
|
+
|
|
|
+ featureMap.put(prefix1 + "_" + prefix2 + "_" + "click", click)
|
|
|
+ featureMap.put(prefix1 + "_" + prefix2 + "_" + "conver_x_log_view", conver * RankExtractorFeature_20240530.calLog(view))
|
|
|
+ featureMap.put(prefix1 + "_" + prefix2 + "_" + "conver_x_ctcvr", conver * f2)
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+ val c1: JSONObject = if (!mateFeature.containsKey("alg_mid_feature_ad_action")) new JSONObject() else
|
|
|
+ mateFeature.getJSONObject("alg_mid_feature_ad_action")
|
|
|
+
|
|
|
+ val midActionList = if (c1.containsKey("action") && c1.getString("action").nonEmpty) {
|
|
|
+ c1.getString("action").split(",").map(r => {
|
|
|
+ val rList = r.split(":")
|
|
|
+ (rList(0), (rList(1).toInt, rList(2).toInt, rList(3).toInt, rList(4).toInt, rList(5)))
|
|
|
+ }).sortBy(-_._2._1).toList
|
|
|
+ } else {
|
|
|
+ new ArrayBuffer[(String, (Int, Int, Int, Int, String))]().toList
|
|
|
+ }
|
|
|
+ // u特征
|
|
|
+ val viewAll = midActionList.size.toDouble
|
|
|
+ val clickAll = midActionList.map(_._2._2).sum.toDouble
|
|
|
+ val converAll = midActionList.map(_._2._3).sum.toDouble
|
|
|
+ val incomeAll = midActionList.map(_._2._4).sum.toDouble
|
|
|
+ featureMap.put("viewAll", viewAll)
|
|
|
+ featureMap.put("clickAll", clickAll)
|
|
|
+ featureMap.put("converAll", converAll)
|
|
|
+ featureMap.put("incomeAll", incomeAll)
|
|
|
+ featureMap.put("ctr_all", RankExtractorFeature_20240530.calDiv(clickAll, viewAll))
|
|
|
+ featureMap.put("ctcvr_all", RankExtractorFeature_20240530.calDiv(converAll, viewAll))
|
|
|
+ featureMap.put("cvr_all", RankExtractorFeature_20240530.calDiv(clickAll, converAll))
|
|
|
+ featureMap.put("ecpm_all", RankExtractorFeature_20240530.calDiv(incomeAll * 1000, viewAll))
|
|
|
+
|
|
|
+ if (c1.containsKey("user_has_conver_1y") && c1.getInteger("user_has_conver_1y") != null) {
|
|
|
+ featureMap.put("user_has_conver_1y", c1.getInteger("user_has_conver_1y"))
|
|
|
+ }
|
|
|
+
|
|
|
+ // ui特征
|
|
|
+ val midTimeDiff = scala.collection.mutable.Map[String, Double]()
|
|
|
+ midActionList.foreach {
|
|
|
+ case (cid, (ts_history, click, conver, income, title)) =>
|
|
|
+ if (!midTimeDiff.contains("timediff_view_" + cid)) {
|
|
|
+ midTimeDiff.put("timediff_view_" + cid, 1.0 / ((ts - ts_history).toDouble / 3600.0 / 24.0))
|
|
|
+ }
|
|
|
+ if (!midTimeDiff.contains("timediff_click_" + cid) && click > 0) {
|
|
|
+ midTimeDiff.put("timediff_click_" + cid, 1.0 / ((ts - ts_history).toDouble / 3600.0 / 24.0))
|
|
|
+ }
|
|
|
+ if (!midTimeDiff.contains("timediff_conver_" + cid) && conver > 0) {
|
|
|
+ midTimeDiff.put("timediff_conver_" + cid, 1.0 / ((ts - ts_history).toDouble / 3600.0 / 24.0))
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ val midActionStatic = scala.collection.mutable.Map[String, Double]()
|
|
|
+ midActionList.foreach {
|
|
|
+ case (cid, (ts_history, click, conver, income, title)) =>
|
|
|
+ midActionStatic.put("actionstatic_view_" + cid, 1.0 + midActionStatic.getOrDefault("actionstatic_view_" + cid, 0.0))
|
|
|
+ midActionStatic.put("actionstatic_click_" + cid, click + midActionStatic.getOrDefault("actionstatic_click_" + cid, 0.0))
|
|
|
+ midActionStatic.put("actionstatic_conver_" + cid, conver + midActionStatic.getOrDefault("actionstatic_conver_" + cid, 0.0))
|
|
|
+ midActionStatic.put("actionstatic_income_" + cid, income + midActionStatic.getOrDefault("actionstatic_income_" + cid, 0.0))
|
|
|
+ }
|
|
|
+
|
|
|
+ val clickCidList = collection.mutable.ListBuffer[String]()
|
|
|
+ val converCidList = collection.mutable.ListBuffer[String]()
|
|
|
+ midActionList.foreach {
|
|
|
+ case (cid, (ts_history, click, conver, income, title)) =>
|
|
|
+ if (click == 1) clickCidList += cid
|
|
|
+ if (conver == 1) converCidList += cid
|
|
|
+ }
|
|
|
+ if (clickCidList.nonEmpty) {
|
|
|
+ featureMap.put("user_cid_click_list", clickCidList.takeRight(50).mkString(","))
|
|
|
+ } else {
|
|
|
+ featureMap.put("user_cid_click_list", "")
|
|
|
+ }
|
|
|
+ if (converCidList.nonEmpty) {
|
|
|
+ featureMap.put("user_cid_conver_list", converCidList.takeRight(50).mkString(","))
|
|
|
+ } else {
|
|
|
+ featureMap.put("user_cid_conver_list", "")
|
|
|
+ }
|
|
|
+ if (midTimeDiff.contains("timediff_view_" + cid)) {
|
|
|
+ featureMap.put("timediff_view", midTimeDiff.getOrDefault("timediff_view_" + cid, 0.0))
|
|
|
+ }
|
|
|
+ if (midTimeDiff.contains("timediff_click_" + cid)) {
|
|
|
+ featureMap.put("timediff_click", midTimeDiff.getOrDefault("timediff_click_" + cid, 0.0))
|
|
|
+ }
|
|
|
+ if (midTimeDiff.contains("timediff_conver_" + cid)) {
|
|
|
+ featureMap.put("timediff_conver", midTimeDiff.getOrDefault("timediff_conver_" + cid, 0.0))
|
|
|
+ }
|
|
|
+ if (midActionStatic.contains("actionstatic_view_" + cid)) {
|
|
|
+ featureMap.put("actionstatic_view", midActionStatic.getOrDefault("actionstatic_view_" + cid, 0.0))
|
|
|
+ }
|
|
|
+ if (midActionStatic.contains("actionstatic_click_" + cid)) {
|
|
|
+ featureMap.put("actionstatic_click", midActionStatic.getOrDefault("actionstatic_click_" + cid, 0.0))
|
|
|
+ }
|
|
|
+ if (midActionStatic.contains("actionstatic_conver_" + cid)) {
|
|
|
+ featureMap.put("actionstatic_conver", midActionStatic.getOrDefault("actionstatic_conver_" + cid, 0.0))
|
|
|
+ }
|
|
|
+ if (midActionStatic.contains("actionstatic_income_" + cid)) {
|
|
|
+ featureMap.put("actionstatic_income", midActionStatic.getOrDefault("actionstatic_income_" + cid, 0.0))
|
|
|
+ }
|
|
|
+ if (midActionStatic.contains("actionstatic_view_" + cid) && midActionStatic.contains("actionstatic_click_" + cid)) {
|
|
|
+ featureMap.put("actionstatic_ctr", RankExtractorFeature_20240530.calDiv(
|
|
|
+ midActionStatic.getOrDefault("actionstatic_click_" + cid, 0.0),
|
|
|
+ midActionStatic.getOrDefault("actionstatic_view_" + cid, 0.0)
|
|
|
+ ))
|
|
|
+ }
|
|
|
+ if (midActionStatic.contains("actionstatic_view_" + cid) && midActionStatic.contains("actionstatic_conver_" + cid)) {
|
|
|
+ featureMap.put("actionstatic_ctcvr", RankExtractorFeature_20240530.calDiv(
|
|
|
+ midActionStatic.getOrDefault("actionstatic_conver_" + cid, 0.0),
|
|
|
+ midActionStatic.getOrDefault("actionstatic_view_" + cid, 0.0)
|
|
|
+ ))
|
|
|
+ }
|
|
|
+ if (midActionStatic.contains("actionstatic_conver_" + cid) && midActionStatic.contains("actionstatic_click_" + cid)) {
|
|
|
+ featureMap.put("actionstatic_cvr", RankExtractorFeature_20240530.calDiv(
|
|
|
+ midActionStatic.getOrDefault("actionstatic_conver_" + cid, 0.0),
|
|
|
+ midActionStatic.getOrDefault("actionstatic_click_" + cid, 0.0)
|
|
|
+ ))
|
|
|
+ }
|
|
|
+
|
|
|
+ val e1: JSONObject = if (!mateFeature.containsKey("alg_mid_feature_return_tags")) new JSONObject() else
|
|
|
+ mateFeature.getJSONObject("alg_mid_feature_return_tags")
|
|
|
+ val e2: JSONObject = if (!mateFeature.containsKey("alg_mid_feature_share_tags")) new JSONObject() else
|
|
|
+ mateFeature.getJSONObject("alg_mid_feature_share_tags")
|
|
|
+ val title = b1.getOrDefault("cidtitle", "").toString
|
|
|
+ if (title.nonEmpty) {
|
|
|
+ for ((en, prefix1) <- List((e1, "e1"), (e2, "e2"))) {
|
|
|
+ for (prefix2 <- List("tags_3d", "tags_7d", "tags_14d")) {
|
|
|
+ if (en.nonEmpty && en.containsKey(prefix2) && en.getString(prefix2).nonEmpty) {
|
|
|
+ val (f1, f2, f3, f4) = funcC34567ForTagsNew(en.getString(prefix2), title)
|
|
|
+ featureMap.put(prefix1 + "_" + prefix2 + "_matchnum", f1)
|
|
|
+ featureMap.put(prefix1 + "_" + prefix2 + "_maxscore", f3)
|
|
|
+ featureMap.put(prefix1 + "_" + prefix2 + "_avgscore", f4)
|
|
|
+
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ if (e1.containsKey("tags_2h") && e1.getString("tags_2h").nonEmpty) {
|
|
|
+ featureMap.put("user_vid_return_tags_2h", e1.getString("tags_2h"))
|
|
|
+ }
|
|
|
+ if (e1.containsKey("tags_1d") && e1.getString("tags_1d").nonEmpty) {
|
|
|
+ featureMap.put("user_vid_return_tags_1d", e1.getString("tags_1d"))
|
|
|
+ }
|
|
|
+ if (e1.containsKey("tags_3d") && e1.getString("tags_3d").nonEmpty) {
|
|
|
+ featureMap.put("user_vid_return_tags_3d", e1.getString("tags_3d"))
|
|
|
+ }
|
|
|
+ if (e1.containsKey("tags_7d") && e1.getString("tags_7d").nonEmpty) {
|
|
|
+ featureMap.put("user_vid_return_tags_7d", e1.getString("tags_7d"))
|
|
|
+ }
|
|
|
+ if (e1.containsKey("tags_14d") && e1.getString("tags_14d").nonEmpty) {
|
|
|
+ featureMap.put("user_vid_return_tags_14d", e1.getString("tags_14d"))
|
|
|
+ }
|
|
|
+
|
|
|
+ if (e2.containsKey("tags_14d") && e2.getString("tags_14d").nonEmpty) {
|
|
|
+ featureMap.put("user_vid_share_tags_1d", e2.getString("tags_1d"))
|
|
|
+ }
|
|
|
+ if (e2.containsKey("tags_14d") && e2.getString("tags_14d").nonEmpty) {
|
|
|
+ featureMap.put("user_vid_share_tags_14d", e2.getString("tags_14d"))
|
|
|
+ }
|
|
|
+
|
|
|
+ val g1: JSONObject = if (!mateFeature.containsKey("mid_return_video_cate")) new JSONObject() else
|
|
|
+ mateFeature.getJSONObject("mid_return_video_cate")
|
|
|
+ val g2: JSONObject = if (!mateFeature.containsKey("mid_share_video_cate")) new JSONObject() else
|
|
|
+ mateFeature.getJSONObject("mid_share_video_cate")
|
|
|
+ if (g1.containsKey("cate1_14d") && g1.getString("cate1_14d").nonEmpty) {
|
|
|
+ featureMap.put("user_vid_return_cate1_14d", g1.getString("cate1_14d"))
|
|
|
+ }
|
|
|
+ if (g1.containsKey("cate2_14d") && g1.getString("cate2_14d").nonEmpty) {
|
|
|
+ featureMap.put("user_vid_return_cate2_14d", g1.getString("cate2_14d"))
|
|
|
+ }
|
|
|
+ if (g2.containsKey("cate1_14d") && g2.getString("cate1_14d").nonEmpty) {
|
|
|
+ featureMap.put("user_vid_share_cate1_14d", g2.getString("cate1_14d"))
|
|
|
+ }
|
|
|
+ if (g2.containsKey("cate2_14d") && g2.getString("cate2_14d").nonEmpty) {
|
|
|
+ featureMap.put("user_vid_share_cate2_14d", g2.getString("cate2_14d"))
|
|
|
+ }
|
|
|
+
|
|
|
+ val h1: JSONObject = if (!mateFeature.containsKey("alg_mid_feature_adver_action")) new JSONObject() else
|
|
|
+ mateFeature.getJSONObject("alg_mid_feature_adver_action")
|
|
|
+ val h2: JSONObject = if (!mateFeature.containsKey("alg_mid_feature_sku_action")) new JSONObject() else
|
|
|
+ mateFeature.getJSONObject("alg_mid_feature_sku_action")
|
|
|
+
|
|
|
+ // 定义时间维度和对应的前缀
|
|
|
+ val timeDimensions = Seq("3d", "7d", "30d")
|
|
|
+ for (dimension <- timeDimensions) {
|
|
|
+ if (h1.containsKey(dimension) && h1.getString(dimension).nonEmpty) {
|
|
|
+ val action = h1.getString(dimension).split(",")
|
|
|
+ if (action.length >= 3) {
|
|
|
+ featureMap.put(s"user_adverid_view_${dimension}", action(0))
|
|
|
+ featureMap.put(s"user_adverid_click_${dimension}", action(1))
|
|
|
+ featureMap.put(s"user_adverid_conver_${dimension}", action(2))
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if (h2.containsKey(dimension) && h2.getString(dimension).nonEmpty) {
|
|
|
+ val action = h2.getString(dimension).split(",")
|
|
|
+ if (action.length >= 3) {
|
|
|
+ featureMap.put(s"user_skuid_view_${dimension}", action(0))
|
|
|
+ featureMap.put(s"user_skuid_click_${dimension}", action(1))
|
|
|
+ featureMap.put(s"user_skuid_conver_${dimension}", action(2))
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+ val d1: JSONObject = if (!mateFeature.containsKey("alg_cid_feature_vid_cf")) new JSONObject() else
|
|
|
+ mateFeature.getJSONObject("alg_cid_feature_vid_cf")
|
|
|
+ val d2: JSONObject = if (!mateFeature.containsKey("alg_cid_feature_vid_cf_rank")) new JSONObject() else
|
|
|
+ mateFeature.getJSONObject("alg_cid_feature_vid_cf_rank")
|
|
|
+ val d3: JSONObject = if (!mateFeature.containsKey("alg_vid_feature_basic_info")) new JSONObject() else
|
|
|
+ mateFeature.getJSONObject("alg_vid_feature_basic_info")
|
|
|
+
|
|
|
+ if (d1.nonEmpty) {
|
|
|
+ for (prefix <- List("3h", "6h", "12h", "1d", "3d", "7d")) {
|
|
|
+ val view = if (!d1.containsKey("ad_view_" + prefix)) 0D else d1.getIntValue("ad_view_" + prefix).toDouble
|
|
|
+ val click = if (!d1.containsKey("ad_click_" + prefix)) 0D else d1.getIntValue("ad_click_" + prefix).toDouble
|
|
|
+ val conver = if (!d1.containsKey("ad_conversion_" + prefix)) 0D else d1.getIntValue("ad_conversion_" + prefix).toDouble
|
|
|
+ val income = if (!d1.containsKey("ad_income_" + prefix)) 0D else d1.getIntValue("ad_income_" + prefix).toDouble
|
|
|
+ val cpc = if (click == 0) 0D else income / click
|
|
|
+ val f1 = RankExtractorFeature_20240530.divSmooth2(click, view, CTR_SMOOTH_BETA_FACTOR)
|
|
|
+ val f2 = RankExtractorFeature_20240530.divSmooth2(conver, view, CTCVR_SMOOTH_BETA_FACTOR)
|
|
|
+ val f3 = RankExtractorFeature_20240530.divSmooth2(conver, click, CVR_SMOOTH_BETA_FACTOR)
|
|
|
+ val f4 = conver
|
|
|
+ val f5 = RankExtractorFeature_20240530.divSmooth2(click, view, CTR_SMOOTH_BETA_FACTOR) * cpc * 1000
|
|
|
+ featureMap.put("d1_feature" + "_" + prefix + "_" + "ctr", f1)
|
|
|
+ featureMap.put("d1_feature" + "_" + prefix + "_" + "ctcvr", f2)
|
|
|
+ featureMap.put("d1_feature" + "_" + prefix + "_" + "cvr", f3)
|
|
|
+ featureMap.put("d1_feature" + "_" + prefix + "_" + "conver", f4)
|
|
|
+ featureMap.put("d1_feature" + "_" + prefix + "_" + "ecpm", f5)
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ val vidRankMaps = scala.collection.mutable.Map[String, scala.collection.immutable.Map[String, Double]]()
|
|
|
+ if (d2.nonEmpty) {
|
|
|
+ d2.foreach(r => {
|
|
|
+ val key = r._1
|
|
|
+ val value = d2.getString(key).split(",").map(r => {
|
|
|
+ val rList = r.split(":")
|
|
|
+ (rList(0), rList(2).toDouble)
|
|
|
+ }).toMap
|
|
|
+ vidRankMaps.put(key, value)
|
|
|
+ })
|
|
|
+ }
|
|
|
+ for (prefix1 <- List("ctr", "ctcvr", "ecpm")) {
|
|
|
+ for (prefix2 <- List("1d", "3d", "7d", "14d")) {
|
|
|
+ if (vidRankMaps.contains(prefix1 + "_" + prefix2)) {
|
|
|
+ val rank = vidRankMaps(prefix1 + "_" + prefix2).getOrDefault(cid, 0.0)
|
|
|
+ if (rank >= 1.0) {
|
|
|
+ featureMap.put("vid_rank_" + prefix1 + "_" + prefix2, 1.0 / rank)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ if (d3.nonEmpty) {
|
|
|
+ val vTitle = d3.getString("title")
|
|
|
+ featureMap.put("cate1", d3.getOrDefault("merge_first_level_cate", ""))
|
|
|
+ featureMap.put("cate2", d3.getOrDefault("merge_second_level_cate", ""))
|
|
|
+ featureMap.put("title_split", d3.getOrDefault("title_split", ""))
|
|
|
+ }
|
|
|
+ featureMap.put("vid", record.getString("headvideoid"))
|
|
|
+
|
|
|
+ /*
|
|
|
+ 广告
|
|
|
+ sparse：cid adid adverid targeting_conversion
|
|
|
+
|
|
|
+ cpa --> 1个
|
|
|
+ adverid下的 3h 6h 12h 1d 3d 7d 、 ctr ctcvr cvr conver ecpm --> 30个
|
|
|
+ cid下的 3h 6h 12h 1d 3d 7d 、 ctr ctcvr cvr ecpm conver --> 30个
|
|
|
+ 地理//cid下的 3h 6h 12h 1d 3d 7d 、 ctr ctcvr cvr ecpm conver --> 30个
|
|
|
+ app//cid下的 3h 6h 12h 1d 3d 7d 、 ctr ctcvr cvr ecpm conver --> 30个
|
|
|
+ 手机品牌//cid下的 3h 6h 12h 1d 3d 7d 、 ctr ctcvr cvr ecpm conver --> 30个
|
|
|
+ 系统 无数据
|
|
|
+ week//cid下的 7d 14d、 ctr ctcvr cvr ecpm conver --> 10个
|
|
|
+ hour//cid下的 7d 14d、 ctr ctcvr cvr ecpm conver --> 10个
|
|
|
+
|
|
|
+ 用户
|
|
|
+ 用户历史 点击/转化 的title tag；3d 7d 14d； cid的title； 数量/最高分/平均分 --> 18个
|
|
|
+ 用户历史 14d 看过/点过/转化次数/income； ctr cvr ctcvr ecpm； --> 8个
|
|
|
+
|
|
|
+ 用户到cid的ui特征 --> 10个
|
|
|
+ 1/用户最近看过这个cid的时间间隔
|
|
|
+ 1/用户最近点过这个cid的时间间隔
|
|
|
+ 1/用户最近转过这个cid的时间间隔
|
|
|
+ 用户看过这个cid多少次
|
|
|
+ 用户点过这个cid多少次
|
|
|
+ 用户转过这个cid多少次
|
|
|
+ 用户对这个cid花了多少钱
|
|
|
+ 用户对这个cid的ctr ctcvr cvr
|
|
|
+
|
|
|
+ 视频
|
|
|
+ vid//cid下的 3h 6h 12h 1d 3d 7d 、 ctr ctcvr cvr ecpm conver --> 30个
|
|
|
+ vid//cid下的 1d 3d 7d 14d、 ctr ctcvr ecpm 的rank值 倒数 --> 12个
|
|
|
+
|
|
|
+ */
|
|
|
+
|
|
|
+
|
|
|
+ //4 处理label信息。
|
|
|
+ val labels = new JSONObject
|
|
|
+ for (labelKey <- List("ad_is_click", "ad_is_conversion")) {
|
|
|
+ if (!record.isNull(labelKey)) {
|
|
|
+ labels.put(labelKey, record.getString(labelKey))
|
|
|
+ }
|
|
|
+ }
|
|
|
+ //5 处理log key表头。
|
|
|
+ val headvideoid = record.getString("headvideoid")
|
|
|
+ val logKey = (apptype, mid, cid, ts, headvideoid).productIterator.mkString(",")
|
|
|
+ val labelKey = labels.toString()
|
|
|
+ (logKey, labelKey, featureMap)
|
|
|
+ })
|
|
|
+ odpsData
|
|
|
+ }).reduce(_ union _)
|
|
|
+ .map { case (logKey, labelKey, jsons) =>
|
|
|
+ val denseFeatures = scala.collection.mutable.Map[String, Double]()
|
|
|
+ val sparseFeatures = scala.collection.mutable.Map[String, String]()
|
|
|
+ denseFeatureNames.foreach(r => {
|
|
|
+ if (jsons.containsKey(r)) {
|
|
|
+ denseFeatures.put(r, jsons.getDoubleValue(r))
|
|
|
+ }
|
|
|
+ })
|
|
|
+ sparseFeatureNames.foreach(r => {
|
|
|
+ if (jsons.get(r) != null) {
|
|
|
+ sparseFeatures.put(r, jsons.get(r).toString)
|
|
|
+ }
|
|
|
+ })
|
|
|
+ (logKey, labelKey, denseFeatures, sparseFeatures)
|
|
|
+ }
|
|
|
+ .map {
|
|
|
+ case (logKey, labelKey, denseFeatures, sparseFeatures) =>
|
|
|
+ val labelObject = JSON.parseObject(labelKey)
|
|
|
+ val label = labelObject.getOrDefault(whatLabel, "0").toString
|
|
|
+ val bucketsMap = bucketsMap_br.value
|
|
|
+ var resultMap = denseFeatures.collect {
|
|
|
+ case (name, score) if !filterNames.exists(name.contains) && score > 1E-8 =>
|
|
|
+ val value = if (bucketsMap.contains(name)) {
|
|
|
+ val (bucketsNum, buckets) = bucketsMap(name)
|
|
|
+ 1.0 / bucketsNum * (ExtractorUtils.findInsertPosition(buckets, score).toDouble + 1.0)
|
|
|
+ } else {
|
|
|
+ score
|
|
|
+ }
|
|
|
+ name -> value.toString
|
|
|
+ }.toMap
|
|
|
+ sparseFeatures.foreach(kv => {
|
|
|
+ resultMap += (kv._1 -> kv._2)
|
|
|
+ })
|
|
|
+ resultMap += ("has_click" -> labelObject.getString("ad_is_click"))
|
|
|
+ resultMap += ("has_conversion" -> labelObject.getString("ad_is_conversion"))
|
|
|
+ resultMap += ("logkey" -> logKey)
|
|
|
+ resultMap
|
|
|
+ }.coalesce(128)
|
|
|
+
|
|
|
+ val partition = s"dt=$dt"
|
|
|
+ if (outputTable2.isEmpty) {
|
|
|
+ odpsOps.saveToTable(project, outputTable, partition, recordRdd, write, defaultCreate = true, overwrite = true)
|
|
|
+ } else {
|
|
|
+ // 固定seed以保证可重入
|
|
|
+ val splitRdds = recordRdd.randomSplit(Array(splitRate, 1 - splitRate), seed = dt.toLong)
|
|
|
+ odpsOps.saveToTable(project, outputTable, partition, splitRdds(0), write, defaultCreate = true, overwrite = true)
|
|
|
+ odpsOps.saveToTable(project, outputTable2, partition, splitRdds(1), write, defaultCreate = true, overwrite = true)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ def write(map: Map[String, String], record: Record, schema: TableSchema): Unit = {
|
|
|
+ for ((columnName, value) <- map) {
|
|
|
+ try {
|
|
|
+ // 查找列名在表结构中的索引
|
|
|
+ val columnIndex = schema.getColumnIndex(columnName.toLowerCase)
|
|
|
+ // 获取列的类型
|
|
|
+ val columnType = schema.getColumn(columnIndex).getTypeInfo
|
|
|
+ try {
|
|
|
+ columnType.getTypeName match {
|
|
|
+ case "STRING" =>
|
|
|
+ record.setString(columnIndex, value)
|
|
|
+ case "BIGINT" =>
|
|
|
+ record.setBigint(columnIndex, value.toLong)
|
|
|
+ case "DOUBLE" =>
|
|
|
+ record.setDouble(columnIndex, value.toDouble)
|
|
|
+ case "BOOLEAN" =>
|
|
|
+ record.setBoolean(columnIndex, value.toBoolean)
|
|
|
+ case other =>
|
|
|
+ throw new IllegalArgumentException(s"Unsupported column type: $other")
|
|
|
+ }
|
|
|
+ } catch {
|
|
|
+ case e: NumberFormatException =>
|
|
|
+ println(s"Error converting value $value to type ${columnType.getTypeName} for column $columnName: ${e.getMessage}")
|
|
|
+ case e: Exception =>
|
|
|
+ println(s"Unexpected error writing value $value to column $columnName: ${e.getMessage}")
|
|
|
+ }
|
|
|
+ } catch {
|
|
|
+ case e: IllegalArgumentException => {
|
|
|
+ println(e.getMessage)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+ def func(record: Record, schema: TableSchema): Record = {
|
|
|
+ record
|
|
|
+ }
|
|
|
+
|
|
|
+ def funcC34567ForTagsNew(tags: String, title: String): Tuple4[Double, String, Double, Double] = {
|
|
|
+ // 匹配数量 匹配词 语义最高相似度分 语义平均相似度分
|
|
|
+ val tagsList = tags.split(",")
|
|
|
+ var d1 = 0.0
|
|
|
+ val d2 = new ArrayBuffer[String]()
|
|
|
+ var d3 = 0.0
|
|
|
+ var d4 = 0.0
|
|
|
+ for (tag <- tagsList) {
|
|
|
+ if (title.contains(tag)) {
|
|
|
+ d1 = d1 + 1.0
|
|
|
+ d2.add(tag)
|
|
|
+ }
|
|
|
+ val score = SimilarityUtils.word2VecSimilarity(tag, title)
|
|
|
+ d3 = if (score > d3) score else d3
|
|
|
+ d4 = d4 + score
|
|
|
+ }
|
|
|
+ d4 = if (tagsList.nonEmpty) d4 / tagsList.size else d4
|
|
|
+ (d1, d2.mkString(","), d3, d4)
|
|
|
+ }
|
|
|
+}
|
