|
|
@@ -0,0 +1,176 @@
|
|
|
+package com.aliyun.odps.spark.examples.makedata_ad
|
|
|
+
|
|
|
+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_ad_33_bucketData_20240729_copy_zheng {
|
|
|
+ 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 readPath = param.getOrElse("readPath", "/dw/recommend/model/31_ad_sample_data/")
|
|
|
+ val savePath = param.getOrElse("savePath", "/dw/recommend/model/33_ad_train_data/")
|
|
|
+ val beginStr = param.getOrElse("beginStr", "20240620")
|
|
|
+ val endStr = param.getOrElse("endStr", "20240620")
|
|
|
+ val repartition = param.getOrElse("repartition", "100").toInt
|
|
|
+ val filterNames = param.getOrElse("filterNames", "").split(",").toSet
|
|
|
+ val whatLabel = param.getOrElse("whatLabel", "ad_is_conversion")
|
|
|
+
|
|
|
+ val loader = getClass.getClassLoader
|
|
|
+
|
|
|
+ val resourceUrlBucket = loader.getResource("20240718_ad_bucket_688.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)
|
|
|
+
|
|
|
+ val dateRange = MyDateUtils.getDateRange(beginStr, endStr)
|
|
|
+ val cidCountMap = scala.collection.mutable.Map[String, Int]()
|
|
|
+ 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)
|
|
|
+ !Set("12", "13").contains(apptype)
|
|
|
+ }.filter {
|
|
|
+ case (logKey, labelKey, features) =>
|
|
|
+ var key = ""
|
|
|
+ for (elem <- features) {
|
|
|
+ if (elem._1.contains("cid_")) {
|
|
|
+ key = elem._1
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ if (key.equals("cid_3319")) {
|
|
|
+ true
|
|
|
+ } else if (key.equals("cid_3024")) {
|
|
|
+ // 创建一个Random实例
|
|
|
+ val rand = new Random()
|
|
|
+
|
|
|
+ // 生成一个0到1之间的随机浮点数
|
|
|
+ val randomDouble = rand.nextDouble()
|
|
|
+
|
|
|
+ randomDouble < 0.01
|
|
|
+ } else {
|
|
|
+ false
|
|
|
+ }
|
|
|
+ }.flatMap {
|
|
|
+ case (logKey, labelKey, features) =>
|
|
|
+ var key = ""
|
|
|
+ for (elem <- features) {
|
|
|
+ if (elem._1.contains("cid_")) {
|
|
|
+ key = elem._1
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if (key.equals("cid_3319")) {
|
|
|
+ val whatLabel = JSON.parseObject(labelKey).getOrDefault(whatLabel, "0").toString
|
|
|
+ if (!whatLabel.equals("0")) {
|
|
|
+ Seq(
|
|
|
+ (logKey, labelKey, features),
|
|
|
+ (logKey, labelKey, features),
|
|
|
+ (logKey, labelKey, features),
|
|
|
+ (logKey, labelKey, features),
|
|
|
+ (logKey, labelKey, features)
|
|
|
+ )
|
|
|
+ } else {
|
|
|
+ Seq((logKey, labelKey, features))
|
|
|
+ }
|
|
|
+ } else {
|
|
|
+ Seq((logKey, labelKey, features))
|
|
|
+ }
|
|
|
+ }.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)
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+ }
|
|
|
+}
|
