add

qiaojialiang/handle_rov.sh

@@ -1,21 +1,20 @@
 #!/bin/sh
 set -ex
 
-#partitionDt="$(date -d '1 days ago' +%Y%m%d)"
-#partitionHh="14"
-#table='alg_recsys_sample_all_test'
+# 原始数据table name
 table='alg_recsys_sample_all'
-
-beginStr=2024070400
-endStr=2024070423
+# 处理分区配置
+beginStr="$(date -d '1 days ago' +%Y%m%d)"
+endStr="$(date -d '1 days ago' +%Y%m%d)"
+beginHhStr=00
+endHhStr=23
 
 # 0 判断上游表是否生产完成，最长等待到12点
+# shellcheck disable=SC2039
 source /root/anaconda3/bin/activate py37
-max_hour=11
-max_minute=00
-echo "开始校验是否生产完数据，分区信息：beginStr:${beginStr},endStr:${endStr}"
+echo "----------step1------------开始校验是否生产完数据，分区信息：beginStr:${beginStr}${beginHhStr},endStr:${endStr}${endHhStr}"
 while true; do
-  python_return_code=$(python checkHiveDataUtil.py --table ${table} --beginStr ${beginStr} --endStr ${endStr})
+  python_return_code=$(python checkHiveDataUtil.py --table ${table} --beginStr ${beginStr}${beginHhStr} --endStr ${endStr}${endHhStr})
   echo "python 返回值：${python_return_code}"
   if [ $python_return_code -eq 0 ]; then
     echo "Python程序返回0，校验存在数据，退出循环。"
@@ -25,6 +24,7 @@ while true; do
   sleep 300
   current_hour=$(date +%H)
   current_minute=$(date +%M)
+  # shellcheck disable=SC2039
   if (( current_hour > max_hour || (current_hour == max_hour && current_minute >= max_minute) )); then
     echo "最长等待时间已到，失败:${current_hour}-${current_minute}"
     exit 1
@@ -32,12 +32,49 @@ while true; do
 done
 
 # 1 生产原始数据
-#nohup /opt/apps/SPARK2/spark-2.4.8-hadoop3.2-1.0.8/bin/spark-class2 org.apache.spark.deploy.SparkSubmit \
-#--class com.aliyun.odps.spark.examples.makedata.makedata_13_originData_20240529 \
-#--master yarn --driver-memory 1G --executor-memory 2G --executor-cores 1 --num-executors 16 \
-#./target/spark-examples-1.0.0-SNAPSHOT-shaded.jar \
-#tablePart:64 repartition:32 \
-#beginStr:${beginStr} endStr:${endStr} \
-#savePath:/dw/recommend/model/13_sample_data/ \
-#table:${beginStr} \
-#> p13_2024061600.log 2>&1 &
+echo "----------step2------------开始根据${table}生产原始数据"
+/opt/apps/SPARK2/spark-2.4.8-hadoop3.2-1.0.8/bin/spark-class2 org.apache.spark.deploy.SparkSubmit \
+--class com.aliyun.odps.spark.examples.makedata_qiao.makedata_13_originData_20240705 \
+--master yarn --driver-memory 1G --executor-memory 2G --executor-cores 1 --num-executors 16 \
+./target/spark-examples-1.0.0-SNAPSHOT-shaded.jar \
+tablePart:64 repartition:32 \
+beginStr:${beginStr} endStr:${endStr} \
+savePath:/dw/recommend/model/13_sample_data/ \
+table:${table}
+if [ $? -ne 0 ]; then
+   echo "Spark原始样本生产任务执行失败"
+   exit 1
+else
+    echo "spark原始样本生产执行成功"
+fi
+
+
+# 2 特征值拼接
+/opt/apps/SPARK2/spark-2.4.8-hadoop3.2-1.0.8/bin/spark-class2 org.apache.spark.deploy.SparkSubmit \
+--class com.aliyun.odps.spark.examples.makedata_qiao.makedata_14_valueData_20240705 \
+--master yarn --driver-memory 1G --executor-memory 3G --executor-cores 1 --num-executors 32 \
+./target/spark-examples-1.0.0-SNAPSHOT-shaded.jar \
+readPath:/dw/recommend/model/13_sample_data/ \
+savePath:/dw/recommend/model/14_feature_data/ \
+beginStr:${beginStr} endStr:${endStr} repartition:1000
+if [ $? -ne 0 ]; then
+   echo "Spark特征值拼接处理任务执行失败"
+   exit 1
+else
+   echo "spark特征值拼接处理执行成功"
+fi
+
+# 3 特征分桶
+echo "----------step3------------根据特征分桶生产重打分特征数据"
+/opt/apps/SPARK2/spark-2.4.8-hadoop3.2-1.0.8/bin/spark-class2 org.apache.spark.deploy.SparkSubmit \
+--class com.aliyun.odps.spark.examples.makedata_qiao.makedata_16_bucketData_20240705 \
+--master yarn --driver-memory 2G --executor-memory 4G --executor-cores 1 --num-executors 16 \
+./target/spark-examples-1.0.0-SNAPSHOT-shaded.jar \
+beginStr:${beginStr} endStr:${endStr} repartition:1000
+if [ $? -ne 0 ]; then
+   echo "Spark特征分桶处理任务执行失败"
+   exit 1
+else
+   echo "spark特征分桶处理执行成功"
+fi
+

src/main/scala/com/aliyun/odps/spark/examples/makdir_qiao/makedata_14_valueData_20240705.scala

@@ -0,0 +1,92 @@
+package com.aliyun.odps.spark.examples.makdir_qiao
+
+import com.alibaba.fastjson.JSON
+import com.aliyun.odps.spark.examples.myUtils.{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
+/*
+
+ */
+
+object makedata_14_valueData_20240705 {
+  def main(args: Array[String]): Unit = {
+
+    val spark = SparkSession
+      .builder()
+      .appName(this.getClass.getName)
+      .getOrCreate()
+    val sc = spark.sparkContext
+
+    val loader = getClass.getClassLoader
+    val resourceUrl = loader.getResource("20240608_feature_name.txt")
+    val content =
+      if (resourceUrl != null) {
+        val content = Source.fromURL(resourceUrl).getLines().mkString("\n")
+        Source.fromURL(resourceUrl).close()
+        content
+      } else {
+        ""
+      }
+    println(content)
+    val contentList = content.split("\n")
+      .map(r=> r.replace(" ", "").replaceAll("\n", ""))
+      .filter(r=> r.nonEmpty).toList
+    val contentList_bc = sc.broadcast(contentList)
+
+
+
+    // 1 读取参数
+    val param = ParamUtils.parseArgs(args)
+    val beginStr = param.getOrElse("beginStr", "20230101")
+    val endStr = param.getOrElse("endStr", "20230101")
+    val readPath = param.getOrElse("readPath", "/dw/recommend/model/13_sample_data/")
+    val savePath = param.getOrElse("savePath", "/dw/recommend/model/14_feature_data/")
+    val repartition = param.getOrElse("repartition", "200").toInt
+    val dateRange = MyDateUtils.getDateRange(beginStr, endStr)
+    for (date <- dateRange) {
+      val data = sc.textFile(readPath + "/" + date + "*")
+      val data1 = data.map(r => {
+        val rList = r.split("\t")
+        val logKey = rList(0)
+        val labelKey = rList(1)
+        val featureKey = rList(2)
+        (logKey, labelKey, featureKey)
+      }).filter(r =>
+        r._1.split(",")(6).equals("0")
+      ).mapPartitions(row => {
+        val result = new ArrayBuffer[String]()
+        val contentList = contentList_bc.value
+        row.foreach {
+          case (logKey, labelKey, featureKey) =>
+            val featureJson = JSON.parseObject(featureKey)
+
+            val featureValues = contentList.map(key => {
+              if (featureJson.containsKey(key)) {
+                featureJson.getDouble(key)
+              } else {
+                0.0
+              }
+            })
+            result.add(logKey + "\t" + labelKey + "\t" + featureValues.mkString(","))
+        }
+        result.iterator
+      })
+
+      // 4 保存数据到hdfs
+      val hdfsPath = savePath + "/" + date
+      if (hdfsPath.nonEmpty && hdfsPath.startsWith("/dw/recommend/model/")) {
+        println("删除路径并开始数据写入:" + hdfsPath)
+        MyHdfsUtils.delete_hdfs_path(hdfsPath)
+        data1.coalesce(repartition).saveAsTextFile(hdfsPath, classOf[GzipCodec])
+      } else {
+        println("路径不合法，无法写入:" + hdfsPath)
+      }
+    }
+
+
+  }
+}

src/main/scala/com/aliyun/odps/spark/examples/makdir_qiao/makedata_16_bucketData_20240705.scala

@@ -0,0 +1,127 @@
+package com.aliyun.odps.spark.examples.makdir_qiao
+
+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
+/*
+
+ */
+
+object makedata_16_bucketData_20240705 {
+  def main(args: Array[String]): Unit = {
+
+    val spark = SparkSession
+      .builder()
+      .appName(this.getClass.getName)
+      .getOrCreate()
+    val sc = spark.sparkContext
+
+    val loader = getClass.getClassLoader
+    val resourceUrl = loader.getResource("20240608_feature_name.txt")
+    val content =
+      if (resourceUrl != null) {
+        val content = Source.fromURL(resourceUrl).getLines().mkString("\n")
+        Source.fromURL(resourceUrl).close()
+        content
+      } else {
+        ""
+      }
+    println(content)
+    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 readPath = param.getOrElse("readPath", "/dw/recommend/model/14_feature_data/")
+    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 logKey = rList(0)
+        val labelKey = rList(1)
+        val features = rList(2).split(",").map(_.toDouble)
+        (logKey, labelKey, features)
+      })
+        .filter{
+          case (logKey, labelKey, features) =>
+            val logKeyList = logKey.split(",")
+            val apptype = logKeyList(0)
+            val pagesource = logKeyList(1)
+            Set("0", "4", "5", "21", "3", "6").contains(apptype) && pagesource.endsWith("recommend")
+        }
+        .map{
+          case (logKey, labelKey, features) =>
+            val label = JSON.parseObject(labelKey).getOrDefault("is_return", "0").toString
+            (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)
+        data.repartition(repartition).saveAsTextFile(hdfsPath, classOf[GzipCodec])
+      } else {
+        println("路径不合法，无法写入:" + hdfsPath)
+      }
+    }
+
+
+
+  }
+}