|
|
@@ -0,0 +1,125 @@
|
|
|
+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_recsys_86_nor_sample_20250627 {
|
|
|
+ def main(args: Array[String]): Unit = {
|
|
|
+ // 1. 读取参数
|
|
|
+ val param = ParamUtils.parseArgs(args)
|
|
|
+ val readPath = param.getOrElse("readPath", "/dw/recommend/model/83_origin_data/")
|
|
|
+ val beginStr = param.getOrElse("beginStr", "20250221")
|
|
|
+ val endStr = param.getOrElse("endStr", "20250221")
|
|
|
+ val whatApps = param.getOrElse("whatApps", "0,4,5,21,3,6").split(",").toSet
|
|
|
+ val whatPages = param.getOrElse("whatPages", "详情后沉浸页,回流后沉浸页&内页feed,首页feed,详情页").split(",").toSet
|
|
|
+ val whatLabel = param.getOrElse("whatLabel", "return_n_uv_noself")
|
|
|
+ val fuSampleRate = param.getOrElse("fuSampleRate", "-1.0").toDouble
|
|
|
+ val notUseBucket = param.getOrElse("notUseBucket", "0").toInt
|
|
|
+ val featureNameFile = param.getOrElse("featureName", "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/83_recsys_nor_train_data/")
|
|
|
+
|
|
|
+ val spark = SparkSession
|
|
|
+ .builder()
|
|
|
+ .appName(this.getClass.getName)
|
|
|
+ .getOrCreate()
|
|
|
+ val sc = spark.sparkContext
|
|
|
+
|
|
|
+ val loader = getClass.getClassLoader
|
|
|
+ val featureNameSet = loadFeatureNames(featureNameFile)
|
|
|
+ val featureBucketMap = DataUtils.loadUseFeatureBuckets(loader, notUseBucket, featureBucketFile)
|
|
|
+ val bucketsMap_br = sc.broadcast(featureBucketMap)
|
|
|
+
|
|
|
+ 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 scoresMap = rList(2)
|
|
|
+ val featData = rList(3)
|
|
|
+ (logKey, labelKey, scoresMap, featData)
|
|
|
+ })
|
|
|
+ .filter {
|
|
|
+ case (logKey, labelKey, scoresMap, featData) =>
|
|
|
+ validData(logKey, whatApps, whatPages)
|
|
|
+ }.filter {
|
|
|
+ case (logKey, labelKey, scoresMap, featData) =>
|
|
|
+ val label = DataUtils.parseLabel(labelKey, whatLabel).toDouble
|
|
|
+ label > 0 || new Random().nextDouble() <= fuSampleRate
|
|
|
+ }
|
|
|
+ .map {
|
|
|
+ case (logKey, labelKey, scoresMap, featData) =>
|
|
|
+ val label = DataUtils.parseLabel(labelKey, whatLabel).toDouble
|
|
|
+ val features = DataUtils.parseFeature(featData)
|
|
|
+ (logKey, label, scoresMap, features)
|
|
|
+ }
|
|
|
+ .mapPartitions(row => {
|
|
|
+ val result = new ArrayBuffer[String]()
|
|
|
+ val bucketsMap = bucketsMap_br.value
|
|
|
+ row.foreach {
|
|
|
+ case (logKey, label, scoresMap, features) =>
|
|
|
+ val featuresBucket = DataUtils.bucketFeature(featureNameSet, bucketsMap, features)
|
|
|
+ result.add(logKey + "\t" + label + "\t" + scoresMap + "\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)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private def recommendFlow(flowPool: String): Boolean = {
|
|
|
+ if (flowPool.isEmpty || flowPool.endsWith("#1")) {
|
|
|
+ return true
|
|
|
+ }
|
|
|
+ false
|
|
|
+ }
|
|
|
+
|
|
|
+ private def validData(logKey: String, whatApps: Set[String], whatPages: Set[String]): Boolean = {
|
|
|
+ // apptype, page, pagesource, recommendpagetype, flowpool, abcode, mid, vid, level, ts
|
|
|
+ val cells = logKey.split(",")
|
|
|
+ val apptype = cells(0)
|
|
|
+ val page = cells(1)
|
|
|
+ //val pagesource = cells(2)
|
|
|
+ val recommendpagetype = cells(3)
|
|
|
+ val flowpool = cells(4)
|
|
|
+ if (whatApps.contains(apptype)) {
|
|
|
+ if (recommendFlow(flowpool)) {
|
|
|
+ if (whatPages.contains(page)) {
|
|
|
+ return true
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ false
|
|
|
+ }
|
|
|
+
|
|
|
+ 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
|
|
|
+ }
|
|
|
+}
|
jch