|
|
@@ -0,0 +1,133 @@
|
|
|
+package com.aliyun.odps.spark.examples.makedata_recsys_r_rate
|
|
|
+
|
|
|
+import com.alibaba.fastjson.JSON
|
|
|
+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
|
|
|
+
|
|
|
+import scala.collection.JavaConversions._
|
|
|
+import scala.collection.mutable.ArrayBuffer
|
|
|
+import scala.io.Source
|
|
|
+import scala.util.Random
|
|
|
+
|
|
|
+object makedata_recsys_61_nor_sample_20241209 {
|
|
|
+ def main(args: Array[String]): Unit = {
|
|
|
+
|
|
|
+ // 1 读取参数
|
|
|
+ val param = ParamUtils.parseArgs(args)
|
|
|
+ val readPath = param.getOrElse("readPath", "/dw/recommend/model/61_origin_data/")
|
|
|
+ val savePath = param.getOrElse("savePath", "/dw/recommend/model/61_recsys_nor_train_data/")
|
|
|
+ val beginStr = param.getOrElse("beginStr", "20241210")
|
|
|
+ val endStr = param.getOrElse("endStr", "20241210")
|
|
|
+ val repartition = param.getOrElse("repartition", "100").toInt
|
|
|
+ val filterNames = param.getOrElse("filterNames", "XXXXXXXXXX").split(",").filter(_.nonEmpty).toSet
|
|
|
+ val whatLabel = param.getOrElse("whatLabel", "total_return_uv_new")
|
|
|
+ val whatApps = param.getOrElse("whatApps", "0,4,5,21,3,6").split(",").toSet
|
|
|
+ val fuSampleRate = param.getOrElse("fuSampleRate", "-1.0").toDouble
|
|
|
+ val fileName = param.getOrElse("fileName", "20241209_recsys_nor_bucket.txt")
|
|
|
+
|
|
|
+ val spark = SparkSession
|
|
|
+ .builder()
|
|
|
+ .appName(this.getClass.getName)
|
|
|
+ .getOrCreate()
|
|
|
+ val sc = spark.sparkContext
|
|
|
+
|
|
|
+ val loader = getClass.getClassLoader
|
|
|
+
|
|
|
+ val resourceUrlBucket = loader.getResource(fileName)
|
|
|
+ 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)
|
|
|
+
|
|
|
+
|
|
|
+ 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 logKey = rList(0)
|
|
|
+ val labelKey = rList(1)
|
|
|
+ val jsons = JSON.parseObject(rList(2))
|
|
|
+ val features = scala.collection.mutable.Map[String, Double]()
|
|
|
+ jsons.foreach(r => {
|
|
|
+ features.put(r._1, jsons.getDoubleValue(r._1))
|
|
|
+ })
|
|
|
+ (logKey, labelKey, features)
|
|
|
+ })
|
|
|
+ .filter {
|
|
|
+ case (logKey, labelKey, features) =>
|
|
|
+ val logKeyList = logKey.split(",")
|
|
|
+ val apptype = logKeyList(0)
|
|
|
+ val pagesource = logKeyList(1)
|
|
|
+ whatApps.contains(apptype) && pagesource.endsWith("recommend")
|
|
|
+ }.filter {
|
|
|
+ case (logKey, labelKey, features) =>
|
|
|
+ val label = JSON.parseObject(labelKey).getOrDefault(whatLabel, "0").toString.toInt
|
|
|
+ label > 0 || new Random().nextDouble() <= fuSampleRate
|
|
|
+ }
|
|
|
+ .map {
|
|
|
+ case (logKey, labelKey, features) =>
|
|
|
+ val label = JSON.parseObject(labelKey).getOrDefault(whatLabel, "0").toString
|
|
|
+ (label, features)
|
|
|
+ }
|
|
|
+ .mapPartitions(row => {
|
|
|
+ val result = new ArrayBuffer[String]()
|
|
|
+ val bucketsMap = bucketsMap_br.value
|
|
|
+ row.foreach {
|
|
|
+ case (label, features) =>
|
|
|
+ val featuresBucket = features.map {
|
|
|
+ case (name, score) =>
|
|
|
+ var ifFilter = false
|
|
|
+ if (filterNames.nonEmpty) {
|
|
|
+ filterNames.foreach(r => if (!ifFilter && name.contains(r)) {
|
|
|
+ ifFilter = true
|
|
|
+ })
|
|
|
+ }
|
|
|
+ if (ifFilter) {
|
|
|
+ ""
|
|
|
+ } else {
|
|
|
+ if (score > 1E-8) {
|
|
|
+ if (bucketsMap.contains(name)) {
|
|
|
+ val (bucketsNum, buckets) = bucketsMap(name)
|
|
|
+ val scoreNew = 1.0 / bucketsNum * (ExtractorUtils.findInsertPosition(buckets, score).toDouble + 1.0)
|
|
|
+ name + ":" + scoreNew.toString
|
|
|
+ } else {
|
|
|
+ name + ":" + score.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)
|
|
|
+ data.repartition(repartition).saveAsTextFile(hdfsPath, classOf[GzipCodec])
|
|
|
+ } else {
|
|
|
+ println("路径不合法，无法写入:" + hdfsPath)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+}
|
jch