|
|
@@ -0,0 +1,123 @@
|
|
|
+package com.aliyun.odps.spark.examples.makedata_recsys_r_rate
|
|
|
+
|
|
|
+import com.aliyun.odps.spark.examples.myUtils.{DataUtils, MyDateUtils, MyHdfsUtils, ParamUtils}
|
|
|
+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_display_ad_sample_20251218 {
|
|
|
+ def main(args: Array[String]): Unit = {
|
|
|
+ // 1. 读取参数
|
|
|
+ val param = ParamUtils.parseArgs(args)
|
|
|
+ val readPath = param.getOrElse("readPath", "/dw/recommend/model/ad_display/data")
|
|
|
+ val beginStr = param.getOrElse("beginStr", "20251216")
|
|
|
+ val endStr = param.getOrElse("endStr", "20251216")
|
|
|
+ val whatLabel = param.getOrElse("whatLabel", "r1_uv")
|
|
|
+ val fuSampleRate = param.getOrElse("fuSampleRate", "0.1").toDouble
|
|
|
+ val notUseBucket = param.getOrElse("notUseBucket", "1").toInt
|
|
|
+ val featureFile = param.getOrElse("featureFile", "20241209_recsys_nor_name.txt")
|
|
|
+ val featureBucketFile = param.getOrElse("featureBucket", "20241209_recsys_nor_bucket.txt")
|
|
|
+ val repartition = param.getOrElse("repartition", "100").toInt
|
|
|
+ val savePath = param.getOrElse("savePath", "/dw/recommend/model/ad_display/sample")
|
|
|
+
|
|
|
+ // 2. content
|
|
|
+ val spark = SparkSession
|
|
|
+ .builder()
|
|
|
+ .appName(this.getClass.getName)
|
|
|
+ .getOrCreate()
|
|
|
+ val sc = spark.sparkContext
|
|
|
+
|
|
|
+ // 3. 处理数据
|
|
|
+ val featureSet = loadFeatureNames(featureFile)
|
|
|
+ val featureBucketMap = loadUseFeatureBuckets(notUseBucket, featureBucketFile)
|
|
|
+ val bucketsMap_br = sc.broadcast(featureBucketMap)
|
|
|
+ val dateRange = MyDateUtils.getDateRange(beginStr, endStr)
|
|
|
+ for (dt <- dateRange) {
|
|
|
+ val partition = "%s".format(dt)
|
|
|
+ println("开始执行:" + partition)
|
|
|
+ val data = sc.textFile(readPath + "/" + partition + "*").map(row => {
|
|
|
+ val cells = row.split("\t")
|
|
|
+ val mid = cells(0)
|
|
|
+ val labels = cells(1)
|
|
|
+ val featData = cells(2)
|
|
|
+ (mid, labels, featData)
|
|
|
+ })
|
|
|
+ .filter {
|
|
|
+ case (mid, labels, featData) =>
|
|
|
+ val label = DataUtils.parseLabel(labels, whatLabel).toInt
|
|
|
+ label > 0 || new Random().nextDouble() <= fuSampleRate
|
|
|
+ }
|
|
|
+ .map {
|
|
|
+ case (mid, labels, featData) =>
|
|
|
+ val label = DataUtils.parseLabel(labels, whatLabel).toInt
|
|
|
+ val features = DataUtils.parseFeature(featData)
|
|
|
+ (mid, label, features)
|
|
|
+ }
|
|
|
+ .mapPartitions(row => {
|
|
|
+ val result = new ArrayBuffer[String]()
|
|
|
+ row.foreach {
|
|
|
+ case (mid, label, features) =>
|
|
|
+ val bucketsMap = bucketsMap_br.value
|
|
|
+ val featuresBucket = DataUtils.bucketFeature(featureSet, bucketsMap, features)
|
|
|
+ if (0 == label) {
|
|
|
+ result.add(mid + "\t" + 0 + "\t" + featuresBucket.mkString("\t"))
|
|
|
+ } else {
|
|
|
+ result.add(mid + "\t" + 0 + "\t" + featuresBucket.mkString("\t"))
|
|
|
+ for (_ <- 1 to label) {
|
|
|
+ result.add(mid + "\t" + 1 + "\t" + featuresBucket.mkString("\t"))
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ result.iterator
|
|
|
+ })
|
|
|
+
|
|
|
+ // 4. 保存数据到hdfs
|
|
|
+ val hdfsPath = savePath + "/" + partition
|
|
|
+ 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)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ def loadFeatureNames(nameFile: String): Set[String] = {
|
|
|
+ val buffer = Source.fromFile(nameFile)
|
|
|
+ val names = buffer.getLines().mkString("\n")
|
|
|
+ buffer.close()
|
|
|
+ val featSet = names.split("\n")
|
|
|
+ .map(r => r.replace(" ", "").replaceAll("\n", ""))
|
|
|
+ .filter(r => r.nonEmpty)
|
|
|
+ .toSet
|
|
|
+ println("featSet.size=" + featSet.size)
|
|
|
+ println(featSet)
|
|
|
+ featSet
|
|
|
+ }
|
|
|
+
|
|
|
+ def loadUseFeatureBuckets(notUseBucket: Int, bucketFile: String): Map[String, (Double, Array[Double])] = {
|
|
|
+ if (notUseBucket > 0) {
|
|
|
+ return Map[String, (Double, Array[Double])]()
|
|
|
+ }
|
|
|
+
|
|
|
+ val buffer = Source.fromFile(bucketFile)
|
|
|
+ val lines = buffer.getLines().mkString("\n")
|
|
|
+ buffer.close()
|
|
|
+ val bucketMap = lines.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
|
|
|
+ println("bucketMap.size=" + bucketMap.size)
|
|
|
+ println(bucketMap)
|
|
|
+ bucketMap
|
|
|
+ }
|
|
|
+}
|