推荐样本生产-特征分桶后结果

src/main/java/examples/extractor/ExtractorUtils.java

@@ -118,7 +118,36 @@ public class ExtractorUtils {
         return (int)bucket;
     }
 
+    public static int findInsertPosition(double[] sortedArray, double target) {
+        int low = 0;
+        int high = sortedArray.length - 1;
+
+        while (low <= high) {
+            int mid = low + (high - low) / 2;
+            double midValue = sortedArray[mid];
+
+            if (midValue < target) {
+                low = mid + 1;
+            } else if (midValue > target) {
+                high = mid - 1;
+            } else {
+                // 找到相等的值，尝试在右侧寻找插入点
+                while (mid < sortedArray.length - 1 && sortedArray[mid + 1] == target) {
+                    mid++;
+                }
+                return mid + 1; // 返回当前mid的下一个位置作为插入点
+            }
+        }
+
+        return low; // 返回low作为插入点
+    }
+
     public static void main(String[] args) {
+        double[] sortedArray = {1.0, 2.0, 4.0, 4.0, 6.0};
+        double target = 0.0;
+        System.out.println(findInsertPosition(sortedArray, target));
+
+
 //        System.out.println(ceilLogRate(0.0002));
 //        System.out.println(ceilLogRate(0.01));
 //        System.out.println(ceilLogRate(0.2));
@@ -128,7 +157,7 @@ public class ExtractorUtils {
 //        System.out.println(bucketCnt(500.));
 //        System.out.println(bucketCnt(50000.));
 
-        System.out.println(generateHourStrings("2024011603", 5));
+//        System.out.println(generateHourStrings("2024011603", 5));
 
     }
 

src/main/scala/com/aliyun/odps/spark/examples/makedata/makedata_16_bucketData_20240609.scala

@@ -1,6 +1,7 @@
 package com.aliyun.odps.spark.examples.makedata
 
-import com.aliyun.odps.spark.examples.myUtils.{MyHdfsUtils, ParamUtils}
+import com.aliyun.odps.spark.examples.myUtils.{MyDateUtils, MyHdfsUtils, ParamUtils}
+import examples.extractor.ExtractorUtils
 import org.apache.hadoop.io.compress.GzipCodec
 import org.apache.spark.sql.SparkSession
 
@@ -34,27 +35,78 @@ object makedata_16_bucketData_20240609 {
     val contentList = content.split("\n")
       .map(r=> r.replace(" ", "").replaceAll("\n", ""))
       .filter(r=> r.nonEmpty).toList
+    val contentList_br = sc.broadcast(contentList)
 
+    val resourceUrlBucket = loader.getResource("20240609_bucket_274.txt")
+    val buckets =
+      if (resourceUrlBucket != null) {
+        val buckets = Source.fromURL(resourceUrlBucket).getLines().mkString("\n")
+        Source.fromURL(resourceUrlBucket).close()
+        buckets
+      } else {
+        ""
+      }
+    println(buckets)
+    val bucketsMap = buckets.split("\n")
+      .map(r => r.replace(" ", "").replaceAll("\n", ""))
+      .filter(r => r.nonEmpty)
+      .map(r =>{
+        val rList = r.split("\t")
+        (rList(0), (rList(1).toDouble, rList(2).split(",").map(_.toDouble)))
+      }).toMap
+    val bucketsMap_br = sc.broadcast(bucketsMap)
 
 
     // 1 读取参数
     val param = ParamUtils.parseArgs(args)
-    val partitionPrefix = param.getOrElse("partitionPrefix", "dt=20240607,hh=00")
-    val date = param.getOrElse("date", "20240607")
     val readPath = param.getOrElse("readPath", "/dw/recommend/model/14_feature_data/")
-    val savePath = param.getOrElse("savePath", "/dw/recommend/model/15_bucket_data/")
-    val bucketNum = param.getOrElse("bucketNum", "200").toInt
+    val savePath = param.getOrElse("savePath", "/dw/recommend/model/16_train_data/")
+    val beginStr = param.getOrElse("beginStr", "20240606")
+    val endStr = param.getOrElse("endStr", "20240607")
+    val repartition = param.getOrElse("repartition", "200").toInt
 
+    val dateRange = MyDateUtils.getDateRange(beginStr, endStr)
+    for (date <- dateRange) {
+      println("开始执行:" + date)
+      val data = sc.textFile(readPath + date).map(r=>{
+        val rList = r.split("\t")
+        val label = rList(0)
+        val features = rList(1).split(",").map(_.toDouble)
+        (label, features)
+      }).mapPartitions(row => {
+        val result = new ArrayBuffer[String]()
+        val contentList = contentList_br.value
+        val bucketsMap = bucketsMap_br.value
+        row.foreach{
+          case (label, features) =>
+            val featuresBucket = contentList.indices.map(i =>{
+              val featureName = contentList(i)
+              val score = features(i)
+              if (score > 1E-8){
+                val (bucketNum, buckets) = bucketsMap(featureName)
+                val scoreNew = 1.0 / bucketNum * (ExtractorUtils.findInsertPosition(buckets, score).toDouble + 1.0)
+                featureName + ":" + scoreNew.toString
+              }else{
+                ""
+              }
+            }).filter(_.nonEmpty)
+            result.add(label + "\t" + featuresBucket.mkString("\t"))
+        }
+        result.iterator
+      })
 
-
-    // 4 保存数据到hdfs
-    val hdfsPath = savePath + "/" + date
-    if (hdfsPath.nonEmpty && hdfsPath.startsWith("/dw/recommend/model/")) {
-      println("删除路径并开始数据写入:" + hdfsPath)
-      MyHdfsUtils.delete_hdfs_path(hdfsPath)
-      data3.repartition(1).saveAsTextFile(hdfsPath, classOf[GzipCodec])
-    } else {
-      println("路径不合法，无法写入:" + hdfsPath)
+      // 4 保存数据到hdfs
+      val hdfsPath = savePath + "/" + date
+      if (hdfsPath.nonEmpty && hdfsPath.startsWith("/dw/recommend/model/")) {
+        println("删除路径并开始数据写入:" + hdfsPath)
+        MyHdfsUtils.delete_hdfs_path(hdfsPath)
+        data.repartition(repartition).saveAsTextFile(hdfsPath, classOf[GzipCodec])
+      } else {
+        println("路径不合法，无法写入:" + hdfsPath)
+      }
     }
+
+
+
   }
 }