增加特征提取

src/main/scala/com/aliyun/odps/spark/examples/makedata_ad/v20240718/makedata_ad_33_bucketDataFromOriginToHive_20260810.scala

@@ -0,0 +1,894 @@
+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.{ParamUtils, env}
+import examples.extractor.{ExtractorUtils, RankExtractorFeature_20240530}
+import examples.utils.{DateTimeUtil, SimilarityUtils}
+import org.apache.spark.sql.SparkSession
+
+import java.text.SimpleDateFormat
+import java.time.{Instant, ZoneId, ZonedDateTime}
+import java.util.Calendar
+import scala.collection.JavaConversions._
+import scala.collection.mutable.ArrayBuffer
+import scala.io.Source
+import scala.language.postfixOps
+import scala.util.Random
+
+/**
+ * 20260810: 从 ad_engine_statistics_log_per5min_new 生产特征。
+ * - id 类：从表字段取（abcode/adid/adverid/apptype/cid/mid/pqtid/vid/extractstrategy）
+ * - 其余特征：从 metafeature 解析（含已并入的 j1~j11 / k1~k4）
+ * - customer_id / landing_page_type：metafeature.reqFeature.customerId / landingPageType
+ * - 不产出 label（has_click / has_conversion）
+ * - beginStr/endStr：输入分区时间 yyyyMMddHHmmss（如 20260810135000），按该区间截取 5 分钟分区
+ * - 输出仍按天分区 dt=yyyyMMdd
+ */
+object makedata_ad_33_bucketDataFromOriginToHive_20260810 {
+  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
+    // 分区时间：yyyyMMddHHmmss，也兼容 yyyyMMdd / yyyyMMddHH
+    val beginStr = param.getOrElse("beginStr", "20260810080000")
+    val endStr = param.getOrElse("endStr", "20260810235500")
+    val project = param.getOrElse("project", "loghubods")
+    val inputTable = param.getOrElse("inputTable", "ad_engine_statistics_log_per5min_new")
+    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
+    // 分割样本集的比例，splitRate部分输出至outputTable，补集输出至outputTable2（如果outputTable2不为空）
+    val splitRate = param.getOrElse("splitRate", "0.9").toDouble
+    val maskFeatureRate = param.getOrElse("maskFeatureRate", "0.0").toDouble
+    val bucketFile = param.getOrElse("bucketFile", "20260807_ad_bucket_1112.txt")
+    val flag = param.getOrElse("flag", "0")
+
+    val loader = getClass.getClassLoader
+    val resourceUrlBucket = loader.getResource(bucketFile)
+    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
+    println(bucketsMap.keySet)
+    val bucketsMap_br = sc.broadcast(bucketsMap)
+    val denseFeatureNames = bucketsMap.keySet
+    val lowerCaseDenseFeatureNames = bucketsMap.keySet.map(_.toLowerCase)
+    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", "user_conver_ad_class", "category_name",
+      "material_md5", "user_layer", "user_layer_l6", "user_class", "user_click_ad_class", "user_view_ad_class",
+      "customer", "customer_id", "landing", "landing_page_type", "agent_id", "flag")
+
+
+    // 2 读取odps+表信息
+    val odpsOps = env.getODPS(sc)
+
+    val tableSchema = odpsOps.getTableSchema(project, outputTable, isPartition = false)
+
+    // 检查所有字段，收集非法字段
+    val invalidFields = tableSchema.flatMap { case (fieldName, _) =>
+      // 跳过 has_click / has_conversion / logkey（本脚本不产出 label）
+      if (fieldName != "has_click" && fieldName != "has_conversion" && fieldName != "logkey") {
+        if (!lowerCaseDenseFeatureNames.contains(fieldName) && !sparseFeatureNames.contains(fieldName)) {
+          Some(fieldName) // 收集缺少字段
+        } else {
+          None
+        }
+      } else {
+        None
+      }
+    }.toList
+
+    // 如果存在非法字段，抛出标准异常
+    if (invalidFields.nonEmpty) {
+      throw new IllegalArgumentException(s"缺少字段: ${invalidFields.mkString(", ")}")
+    }
+
+    // 3 按 beginStr~endStr 截取 5 分钟分区，再按天写出
+    val partitionTimes = get5MinPartitionRange(beginStr, endStr).filter { p =>
+      val hh = p.substring(8, 10)
+      filterHours.isEmpty || !filterHours.contains(hh)
+    }
+    println(s"partition range: $beginStr ~ $endStr, count=${partitionTimes.size}")
+    val partitionsByDay = partitionTimes.groupBy(_.substring(0, 8)).toSeq.sortBy(_._1)
+    for ((dt, dayParts) <- partitionsByDay) {
+      val inputPartitions = dayParts.map(p => s"dt=$p")
+      if (inputPartitions.isEmpty) {
+        println(s"skip dt=$dt, no input partitions after filterHours")
+      } else {
+      val recordRdd = inputPartitions.map { partition =>
+          println("read partition:" + partition)
+          val odpsData = odpsOps.readTable(project = project,
+              table = inputTable,
+              partition = partition,
+              transfer = func,
+              numPartition = tablePart)
+            .filter(record => !record.isNull("metafeature") && record.getString("metafeature").nonEmpty)
+            .filter(record => Option(record.getString("extractstrategy")).contains("engine"))
+            .filter(record => {
+              val appType = Option(record.getString("apptype")).getOrElse("")
+              !Set("12", "13").contains(appType)
+            })
+            .filter(record => {
+              val adverId = Option(record.getString("adverid")).getOrElse("")
+              !filterAdverIds.contains(adverId)
+            })
+            .map(record => {
+              // id 类从表字段取；其余从 metafeature 解析
+              val mateFeature: JSONObject = parseJsonObject(record.getString("metafeature"))
+              val featureMap = new JSONObject()
+              val reqFeature: JSONObject = getJsonFromMeta(mateFeature, "reqFeature")
+              val sceneFeature: JSONObject = getJsonFromMeta(mateFeature, "sceneFeature")
+              val b1: JSONObject = getJsonFromMeta(mateFeature, "alg_cid_feature_basic_info")
+              val b2: JSONObject = getJsonFromMeta(mateFeature, "alg_cid_feature_adver_action")
+              val b3: JSONObject = getJsonFromMeta(mateFeature, "alg_cid_feature_cid_action")
+              val b4: JSONObject = getJsonFromMeta(mateFeature, "alg_cid_feature_region_action")
+              val b5: JSONObject = getJsonFromMeta(mateFeature, "alg_cid_feature_app_action")
+              val b6: JSONObject = getJsonFromMeta(mateFeature, "alg_cid_feature_week_action")
+              val b7: JSONObject = getJsonFromMeta(mateFeature, "alg_cid_feature_hour_action")
+              val b8: JSONObject = getJsonFromMeta(mateFeature, "alg_cid_feature_brand_action")
+              val b9: JSONObject = getJsonFromMeta(mateFeature, "alg_cid_feature_weChatVersion_action")
+              // j/k 已并入 metafeature
+              val j1: JSONObject = getJsonFromMeta(mateFeature, "j1_feature")
+              val j2: JSONObject = getJsonFromMeta(mateFeature, "j2_feature")
+              val j3: JSONObject = getJsonFromMeta(mateFeature, "j3_feature")
+              val j4: JSONObject = getJsonFromMeta(mateFeature, "j4_feature")
+              val j5: JSONObject = getJsonFromMeta(mateFeature, "j5_feature")
+              val j6: JSONObject = getJsonFromMeta(mateFeature, "j6_feature")
+              val j7: JSONObject = getJsonFromMeta(mateFeature, "j7_feature")
+              val j8: JSONObject = getJsonFromMeta(mateFeature, "j8_feature")
+              val j9: JSONObject = getJsonFromMeta(mateFeature, "j9_feature")
+              val j10: JSONObject = getJsonFromMeta(mateFeature, "j10_feature")
+              val j11: JSONObject = getJsonFromMeta(mateFeature, "j11_feature")
+              val k1: JSONObject = getJsonFromMeta(mateFeature, "k1_feature")
+              val k2: JSONObject = getJsonFromMeta(mateFeature, "k2_feature")
+              val k3: JSONObject = getJsonFromMeta(mateFeature, "k3_feature")
+              val k4: JSONObject = getJsonFromMeta(mateFeature, "k4_feature")
+
+              // id 类：直接取表字段
+              val cid = Option(record.getString("cid")).getOrElse("")
+              val adid = Option(record.getString("adid")).getOrElse("")
+              val adverId = Option(record.getString("adverid")).getOrElse("")
+              val mid = Option(record.getString("mid")).getOrElse("")
+              val pqtid = Option(record.getString("pqtid")).getOrElse("")
+              val apptype = Option(record.getString("apptype")).getOrElse("")
+              val abcode = Option(record.getString("abcode")).getOrElse("")
+              val vid = Option(record.getString("vid")).getOrElse("")
+              // ts：sceneFeature.ts
+              val tsStr = firstNonEmpty(
+                sceneFeature.getString("ts"),
+                mateFeature.getString("ts"),
+                reqFeature.getString("ts")
+              )
+              val targetingConversion = firstNonEmpty(
+                reqFeature.getString("targeting_conversion"),
+                b1.getString("targeting_conversion")
+              )
+
+              if (tsStr.isEmpty || cid.isEmpty) {
+                None
+              } else {
+                val ts = tsStr.toInt
+                val instant = Instant.ofEpochSecond(ts)
+                val chinaZone = ZoneId.of("Asia/Shanghai")
+                val zonedDateTime = ZonedDateTime.ofInstant(instant, chinaZone)
+                val dayOfTheWeek = zonedDateTime.getDayOfWeek.getValue()
+                val isWeekday = if (dayOfTheWeek <= 5) 1 else 2
+
+                featureMap.put("apptype", apptype)
+                featureMap.put("ts", ts)
+                featureMap.put("mid", mid)
+                featureMap.put("pqtid", pqtid)
+                featureMap.put("targeting_conversion", targetingConversion)
+                if (cid.nonEmpty) featureMap.put("cid", cid)
+                if (adid.nonEmpty) featureMap.put("adid", adid)
+                if (adverId.nonEmpty) featureMap.put("adverid", adverId)
+                if (vid.nonEmpty) featureMap.put("vid", vid)
+
+                featureMap.put("cid_" + cid, idDefaultValue)
+                if (adid.nonEmpty) {
+                  featureMap.put("adid_" + adid, idDefaultValue)
+                }
+                if (adverId.nonEmpty) {
+                  featureMap.put("adverid_" + adverId, idDefaultValue)
+                }
+                if (targetingConversion.nonEmpty) {
+                  featureMap.put("targeting_conversion_" + targetingConversion, idDefaultValue)
+                }
+                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("|"))
+                }
+
+                // customer / landing：reqFeature.customerId / landingPageType
+                val customerId = Option(reqFeature.getString("customerId")).getOrElse("")
+                if (customerId.nonEmpty) {
+                  featureMap.put("customer", customerId)
+                  featureMap.put("customer_id", customerId)
+                }
+                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)
+
+                if (apptype.nonEmpty) {
+                  featureMap.put("apptype_" + apptype, idDefaultValue)
+                }
+
+                if (abcode.nonEmpty) {
+                  featureMap.put("abcode_" + abcode, idDefaultValue)
+                }
+
+                // 场景/画像等仍从 reqFeature 取；id 类已用表字段覆盖
+                val reqFeatureKeys = List(
+                  "profession", "region",
+                  "city", "is_first_layer", "root_source_scene",
+                  "root_source_channel", "brand", "category_name", "material_md5"
+                )
+                reqFeatureKeys.foreach { key =>
+                  val value = reqFeature.getString(key)
+                  if (value != null && value.nonEmpty) {
+                    featureMap.put(key, value)
+                  }
+                }
+                val landingPageType = Option(reqFeature.getString("landingPageType")).getOrElse("")
+                if (landingPageType.nonEmpty) {
+                  featureMap.put("landing", landingPageType)
+                  featureMap.put("landing_page_type", landingPageType)
+                }
+                if (reqFeature.containsKey("agentId") && reqFeature.getString("agentId").nonEmpty) {
+                  featureMap.put("agent_id", reqFeature.getString("agentId"))
+                }
+                if (reqFeature.containsKey("layer_l4")) {
+                  featureMap.put("user_layer", reqFeature.getString("layer_l4"))
+                }
+                if (reqFeature.containsKey("layer_l6") && reqFeature.getString("layer_l6").nonEmpty) {
+                  featureMap.put("user_layer_l6", reqFeature.getString("layer_l6"))
+                } else {
+                  featureMap.put("user_layer_l6", "无曝光")
+                }
+                if (reqFeature.containsKey("clazz_l4")) {
+                  featureMap.put("user_class", reqFeature.getString("clazz_l4"))
+                }
+                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
+                    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 = getJsonFromMeta(mateFeature, "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
+                }
+                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"))
+                }
+                if (c1.containsKey("user_conver_ad_class") && c1.getString("user_conver_ad_class") != null) {
+                  featureMap.put("user_conver_ad_class", c1.getString("user_conver_ad_class"))
+                }
+                if (c1.containsKey("user_click_ad_class") && c1.getString("user_click_ad_class") != null) {
+                  featureMap.put("user_click_ad_class", c1.getString("user_click_ad_class"))
+                }
+                if (c1.containsKey("user_view_ad_class") && c1.getString("user_view_ad_class") != null) {
+                  featureMap.put("user_view_ad_class", c1.getString("user_view_ad_class"))
+                }
+
+                val midTimeDiff = scala.collection.mutable.Map[String, Double]()
+                midActionList.foreach {
+                  case (cidHis, (ts_history, click, conver, income, title)) =>
+                    if (!midTimeDiff.contains("timediff_view_" + cidHis)) {
+                      midTimeDiff.put("timediff_view_" + cidHis, 1.0 / ((ts - ts_history).toDouble / 3600.0 / 24.0))
+                    }
+                    if (!midTimeDiff.contains("timediff_click_" + cidHis) && click > 0) {
+                      midTimeDiff.put("timediff_click_" + cidHis, 1.0 / ((ts - ts_history).toDouble / 3600.0 / 24.0))
+                    }
+                    if (!midTimeDiff.contains("timediff_conver_" + cidHis) && conver > 0) {
+                      midTimeDiff.put("timediff_conver_" + cidHis, 1.0 / ((ts - ts_history).toDouble / 3600.0 / 24.0))
+                    }
+                }
+
+                val midActionStatic = scala.collection.mutable.Map[String, Double]()
+                midActionList.foreach {
+                  case (cidHis, (ts_history, click, conver, income, title)) =>
+                    midActionStatic.put("actionstatic_view_" + cidHis, 1.0 + midActionStatic.getOrDefault("actionstatic_view_" + cidHis, 0.0))
+                    midActionStatic.put("actionstatic_click_" + cidHis, click + midActionStatic.getOrDefault("actionstatic_click_" + cidHis, 0.0))
+                    midActionStatic.put("actionstatic_conver_" + cidHis, conver + midActionStatic.getOrDefault("actionstatic_conver_" + cidHis, 0.0))
+                    midActionStatic.put("actionstatic_income_" + cidHis, income + midActionStatic.getOrDefault("actionstatic_income_" + cidHis, 0.0))
+                }
+
+                val clickCidList = collection.mutable.ListBuffer[String]()
+                val converCidList = collection.mutable.ListBuffer[String]()
+                midActionList.foreach {
+                  case (cidHis, (ts_history, click, conver, income, title)) =>
+                    if (click == 1) clickCidList += cidHis
+                    if (conver == 1) converCidList += cidHis
+                }
+                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 = getJsonFromMeta(mateFeature, "alg_mid_feature_return_tags")
+                val e2: JSONObject = getJsonFromMeta(mateFeature, "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 = getJsonFromMeta(mateFeature, "mid_return_video_cate")
+                val g2: JSONObject = getJsonFromMeta(mateFeature, "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 = getJsonFromMeta(mateFeature, "alg_mid_feature_adver_action")
+                val h2: JSONObject = getJsonFromMeta(mateFeature, "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 = getJsonFromMeta(mateFeature, "alg_cid_feature_vid_cf")
+                val d2: JSONObject = getJsonFromMeta(mateFeature, "alg_cid_feature_vid_cf_rank")
+                val d3: JSONObject = getJsonFromMeta(mateFeature, "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) {
+                  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", ""))
+                }
+
+                if (Random.nextDouble() < maskFeatureRate) {
+                  featureMap.put("cid", "")
+                  featureMap.put("adid", "")
+                  featureMap.put("adverid", "")
+                  featureMap.put("customer", "")
+                }
+                featureMap.put("flag", flag)
+
+                val jList: List[(String, JSONObject, List[String])] = List(
+                  ("j1", j1, List("3h", "3d")),
+                  ("j2", j2, List("3h", "3d")),
+                  ("j3", j3, List("3h", "3d")),
+                  ("j4", j4, List("3h", "3d")),
+                  ("j5", j5, List("3h", "3d")),
+                  ("j6", j6, List("1h", "2h", "3h", "6h", "12h", "1d", "3d", "today", "yesterday")),
+                  ("j7", j7, List("3h", "3d")),
+                  ("j8", j8, List("3h", "3d")),
+                  ("j9", j9, List("3h", "3d")),
+                  ("j10", j10, List("3h", "3d")),
+                  ("j11", j11, List("3h", "3d"))
+                )
+                for ((prefix1, bn, periods) <- jList) {
+                  for (prefix2 <- periods) {
+                    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 kList: List[(String, JSONObject)] = List(
+                  ("k1", k1), ("k2", k2), ("k3", k3), ("k4", k4)
+                )
+                val kPeriods = List("2h", "4h", "6h", "12h", "1d", "3d", "today", "1w")
+                for ((kPrefix, kn) <- kList) {
+                  for (period <- kPeriods) {
+                    val view = if (kn.isEmpty) 0D else kn.getIntValue("ad_view_" + period).toDouble
+                    val click = if (kn.isEmpty) 0D else kn.getIntValue("ad_click_" + period).toDouble
+                    val eventJson = parseEventJson(kn, "event_" + period)
+                    val conver =
+                      if (eventJson.isEmpty || targetingConversion.isEmpty) 0D
+                      else eventJson.getIntValue(targetingConversion + "_" + period).toDouble
+                    val ctr = RankExtractorFeature_20240530.divSmooth2(click, view, CTR_SMOOTH_BETA_FACTOR)
+                    val cvr = RankExtractorFeature_20240530.divSmooth2(conver, click, CVR_SMOOTH_BETA_FACTOR)
+                    val ctvr = RankExtractorFeature_20240530.divSmooth2(conver, view, CTCVR_SMOOTH_BETA_FACTOR)
+                    val featPrefix = kPrefix + "_" + period
+                    featureMap.put(featPrefix + "_view", view)
+                    featureMap.put(featPrefix + "_click", click)
+                    featureMap.put(featPrefix + "_conver", conver)
+                    featureMap.put(featPrefix + "_ctr", ctr)
+                    featureMap.put(featPrefix + "_cvr", cvr)
+                    featureMap.put(featPrefix + "_ctvr", ctvr)
+                  }
+                }
+
+                // 不产出 label；logKey 用 pqtid（pqtid 同时作为 sparse 特征保留）
+                val logKey = pqtid
+                Some((logKey, featureMap))
+              }
+            }).collect { case Some(v) => v }
+            .map { case (logKey, 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, denseFeatures, sparseFeatures)
+            }
+            .map {
+              case (logKey, denseFeatures, sparseFeatures) =>
+                val bucketsMap = bucketsMap_br.value
+                def keepDense(name: String): Boolean =
+                  name.startsWith("k1_") || name.startsWith("k2_") || name.startsWith("k3_") || name.startsWith("k4_") ||
+                    !filterNames.exists(name.contains)
+                var resultMap = denseFeatures.collect {
+                  case (name, score) if keepDense(name) && 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 += ("logkey" -> logKey)
+                resultMap
+            }
+          odpsData
+        }.reduce(_ union _)
+        .coalesce(128)
+
+      val outPartition = s"dt=$dt"
+      if (outputTable2.isEmpty) {
+        odpsOps.saveToTable(project, outputTable, outPartition, recordRdd, write, defaultCreate = true, overwrite = true)
+      } else {
+        val splitRdds = recordRdd.randomSplit(Array(splitRate, 1 - splitRate), seed = dt.toLong)
+        odpsOps.saveToTable(project, outputTable, outPartition, splitRdds(0), write, defaultCreate = true, overwrite = true)
+        odpsOps.saveToTable(project, outputTable2, outPartition, 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)
+  }
+
+  /** 规范化为 yyyyMMddHHmmss，并向下对齐到 5 分钟。兼容 yyyyMMdd / yyyyMMddHH / yyyyMMddHHmm */
+  def normalizeTo5MinPartition(raw: String, isEnd: Boolean): String = {
+    val s = Option(raw).getOrElse("").trim
+    val padded = s.length match {
+      case 8 => if (isEnd) s + "235500" else s + "000000"
+      case 10 => if (isEnd) s + "5500" else s + "0000"
+      case 12 => s + "00"
+      case 14 => s
+      case _ =>
+        throw new IllegalArgumentException(s"invalid time: $raw, expect yyyyMMddHHmmss (or yyyyMMdd / yyyyMMddHH)")
+    }
+    val min = padded.substring(10, 12).toInt
+    val alignedMin = (min / 5) * 5
+    f"${padded.substring(0, 10)}${alignedMin}%02d00"
+  }
+
+  /** [beginStr, endStr] 闭区间，步长 5 分钟，返回 yyyyMMddHHmmss 列表 */
+  def get5MinPartitionRange(beginStr: String, endStr: String): Seq[String] = {
+    val sdf = new SimpleDateFormat("yyyyMMddHHmmss")
+    val begin = sdf.parse(normalizeTo5MinPartition(beginStr, isEnd = false))
+    val end = sdf.parse(normalizeTo5MinPartition(endStr, isEnd = true))
+    if (begin.after(end)) {
+      throw new IllegalArgumentException(s"beginStr > endStr: $beginStr > $endStr")
+    }
+    val ranges = ArrayBuffer[String]()
+    val cal = Calendar.getInstance()
+    cal.setTime(begin)
+    while (!cal.getTime.after(end)) {
+      ranges += sdf.format(cal.getTime)
+      cal.add(Calendar.MINUTE, 5)
+    }
+    ranges
+  }
+
+  def parseJsonObject(raw: String): JSONObject = {
+    if (raw == null || raw.isEmpty) {
+      new JSONObject()
+    } else {
+      try {
+        val parsed = JSON.parseObject(raw)
+        if (parsed == null) new JSONObject() else parsed
+      } catch {
+        case _: Exception => new JSONObject()
+      }
+    }
+  }
+
+  def getJsonFromMeta(meta: JSONObject, name: String): JSONObject = {
+    if (meta == null || meta.isEmpty || !meta.containsKey(name)) {
+      return new JSONObject()
+    }
+    val value = meta.get(name)
+    if (value == null) {
+      return new JSONObject()
+    }
+    value match {
+      case obj: JSONObject => obj
+      case s: String => parseJsonObject(s)
+      case _ =>
+        try {
+          val parsed = meta.getJSONObject(name)
+          if (parsed == null) parseJsonObject(String.valueOf(value)) else parsed
+        } catch {
+          case _: Exception => parseJsonObject(String.valueOf(value))
+        }
+    }
+  }
+
+  def firstNonEmpty(values: String*): String = {
+    values.find(v => v != null && v.nonEmpty).getOrElse("")
+  }
+
+  def parseEventJson(feature: JSONObject, eventKey: String): JSONObject = {
+    if (feature == null || feature.isEmpty || !feature.containsKey(eventKey)) {
+      return new JSONObject()
+    }
+    val raw = feature.getString(eventKey)
+    if (raw == null || raw.isEmpty) {
+      return new JSONObject()
+    }
+    try {
+      val parsed = JSON.parseObject(raw)
+      if (parsed == null) new JSONObject() else parsed
+    } catch {
+      case _: Exception => new JSONObject()
+    }
+  }
+}