|
|
@@ -10,7 +10,9 @@ import org.apache.spark.rdd.RDD
|
|
|
import org.apache.spark.sql.types.{DataTypes, StructField}
|
|
|
import org.apache.spark.sql.{Dataset, Row, RowFactory, SparkSession}
|
|
|
|
|
|
+import scala.collection.JavaConversions._
|
|
|
import java.util
|
|
|
+import scala.collection.mutable.ArrayBuffer
|
|
|
import scala.io.Source
|
|
|
|
|
|
object train_01_xgb_ad_20240808{
|
|
|
@@ -20,10 +22,24 @@ object train_01_xgb_ad_20240808{
|
|
|
.appName(this.getClass.getName)
|
|
|
.getOrCreate()
|
|
|
val sc = spark.sparkContext
|
|
|
-// val features = Array("cpa", "b2_12h_ctr", "b2_12h_ctcvr", "b2_12h_cvr", "b2_12h_conver", "b2_12h_click", "b2_12h_conver*log(view)", "b2_12h_conver*ctcvr", "b2_7d_ctr", "b2_7d_ctcvr", "b2_7d_cvr", "b2_7d_conver", "b2_7d_click", "b2_7d_conver*log(view)", "b2_7d_conver*ctcvr")
|
|
|
+
|
|
|
+ val param = ParamUtils.parseArgs(args)
|
|
|
+ val featureFile = param.getOrElse("featureFile", "20240703_ad_feature_name.txt")
|
|
|
+ val trainPath = param.getOrElse("trainPath", "/dw/recommend/model/33_ad_train_data_v4/20240724")
|
|
|
+ val testPath = param.getOrElse("testPath", "/dw/recommend/model/33_ad_train_data_v4/20240725")
|
|
|
+ val savePath = param.getOrElse("savePath", "/dw/recommend/model/34_ad_predict_data/")
|
|
|
+ val featureFilter = param.getOrElse("featureFilter", "XXXXXX").split(",")
|
|
|
+ val eta = param.getOrElse("eta", "0.01").toDouble
|
|
|
+ val gamma = param.getOrElse("gamma", "0.0").toDouble
|
|
|
+ val max_depth = param.getOrElse("max_depth", "5").toInt
|
|
|
+ val num_round = param.getOrElse("num_round", "100").toInt
|
|
|
+ val num_worker = param.getOrElse("num_worker", "20").toInt
|
|
|
+ val func_object = param.getOrElse("func_object", "binary:logistic")
|
|
|
+ val func_metric = param.getOrElse("func_metric", "auc")
|
|
|
+ val repartition = param.getOrElse("repartition", "20").toInt
|
|
|
|
|
|
val loader = getClass.getClassLoader
|
|
|
- val resourceUrl = loader.getResource("20240703_ad_feature_name.txt")
|
|
|
+ val resourceUrl = loader.getResource(featureFile)
|
|
|
val content =
|
|
|
if (resourceUrl != null) {
|
|
|
val content = Source.fromURL(resourceUrl).getLines().mkString("\n")
|
|
|
@@ -33,20 +49,27 @@ object train_01_xgb_ad_20240808{
|
|
|
""
|
|
|
}
|
|
|
println(content)
|
|
|
+
|
|
|
val features = content.split("\n")
|
|
|
.map(r => r.replace(" ", "").replaceAll("\n", ""))
|
|
|
- .filter(r => r.nonEmpty)
|
|
|
-
|
|
|
+ .filter(r => r.nonEmpty || !featureFilter.contains(r))
|
|
|
+ println("features.size=" + features.length)
|
|
|
|
|
|
- val trainData = createData(
|
|
|
- sc.textFile("/dw/recommend/model/33_ad_train_data_v4/20240724"),
|
|
|
+ val trainData = createData4Ad(
|
|
|
+ sc.textFile(trainPath),
|
|
|
features
|
|
|
)
|
|
|
- println("train data size：" + trainData.count())
|
|
|
+ println("zhangbo:train data size：" + trainData.count())
|
|
|
|
|
|
- val fields = Array(
|
|
|
+ var fields = Array(
|
|
|
DataTypes.createStructField("label", DataTypes.IntegerType, true)
|
|
|
+// DataTypes.createStructField("logKey", DataTypes.IntegerType, true)
|
|
|
+
|
|
|
) ++ features.map(f => DataTypes.createStructField(f, DataTypes.DoubleType, true))
|
|
|
+
|
|
|
+ fields = fields ++ Array(
|
|
|
+ DataTypes.createStructField("logKey", DataTypes.StringType, true)
|
|
|
+ )
|
|
|
val schema = DataTypes.createStructType(fields)
|
|
|
val trainDataSet: Dataset[Row] = spark.createDataFrame(trainData, schema)
|
|
|
val vectorAssembler = new VectorAssembler().setInputCols(features).setOutputCol("features")
|
|
|
@@ -56,39 +79,43 @@ object train_01_xgb_ad_20240808{
|
|
|
// "objective" -> "binary:logistic",
|
|
|
// "num_class" -> 3)
|
|
|
val xgbClassifier = new XGBoostClassifier()
|
|
|
- .setEta(0.01f)
|
|
|
+ .setEta(eta)
|
|
|
+ .setGamma(gamma)
|
|
|
.setMissing(0.0f)
|
|
|
- .setMaxDepth(5)
|
|
|
- .setNumRound(1000)
|
|
|
+ .setMaxDepth(max_depth)
|
|
|
+ .setNumRound(num_round)
|
|
|
.setSubsample(0.8)
|
|
|
.setColsampleBytree(0.8)
|
|
|
.setScalePosWeight(1)
|
|
|
- .setObjective("binary:logistic")
|
|
|
- .setEvalMetric("auc")
|
|
|
+ .setObjective(func_object)
|
|
|
+ .setEvalMetric(func_metric)
|
|
|
.setFeaturesCol("features")
|
|
|
.setLabelCol("label")
|
|
|
.setNthread(1)
|
|
|
- .setNumWorkers(1)
|
|
|
+ .setNumWorkers(num_worker)
|
|
|
+ .setSeed(2024)
|
|
|
+ .setMinChildWeight(1)
|
|
|
val model = xgbClassifier.fit(xgbInput)
|
|
|
|
|
|
|
|
|
- val testData = createData(
|
|
|
- sc.textFile("/dw/recommend/model/33_ad_train_data_v4/20240725"),
|
|
|
+ val testData = createData4Ad(
|
|
|
+ sc.textFile(testPath),
|
|
|
features
|
|
|
)
|
|
|
val testDataSet = spark.createDataFrame(testData, schema)
|
|
|
- val testDataSetTrans = vectorAssembler.transform(testDataSet).select("features","label")
|
|
|
+ val testDataSetTrans = vectorAssembler.transform(testDataSet).select("features","label", "logKey")
|
|
|
val predictions = model.transform(testDataSetTrans)
|
|
|
-
|
|
|
- val saveData = predictions.select("label", "prediction", "features", "rawPrediction", "probability").rdd
|
|
|
+// [label, features, probability, prediction, rawPrediction]
|
|
|
+ println("zhangbo:columns:" + predictions.columns.mkString(","))
|
|
|
+ val saveData = predictions.select("label", "rawPrediction", "probability", "logKey").rdd
|
|
|
.map(r =>{
|
|
|
- (r.get(0), r.get(1), r.get(2), r.get(3), r.get(4)).productIterator.mkString("\t")
|
|
|
+ (r.get(0), r.get(1), r.get(2), r.get(3)).productIterator.mkString("\t")
|
|
|
})
|
|
|
- val hdfsPath = "/dw/recommend/model/checkpoint_xgbtest"
|
|
|
+ val hdfsPath = savePath
|
|
|
if (hdfsPath.nonEmpty && hdfsPath.startsWith("/dw/recommend/model/")) {
|
|
|
println("删除路径并开始数据写入:" + hdfsPath)
|
|
|
MyHdfsUtils.delete_hdfs_path(hdfsPath)
|
|
|
- saveData.repartition(100).saveAsTextFile(hdfsPath, classOf[GzipCodec])
|
|
|
+ saveData.repartition(repartition).saveAsTextFile(hdfsPath, classOf[GzipCodec])
|
|
|
} else {
|
|
|
println("路径不合法，无法写入:" + hdfsPath)
|
|
|
}
|
|
|
@@ -101,24 +128,64 @@ object train_01_xgb_ad_20240808{
|
|
|
.setMetricName("areaUnderROC")
|
|
|
val auc = evaluator.evaluate(predictions.select("label", "probability"))
|
|
|
println("zhangbo:auc:" + auc)
|
|
|
+
|
|
|
+ // 统计分cid的分数
|
|
|
+ sc.textFile(hdfsPath).map(r=>{
|
|
|
+ val rList = r.split("\t")
|
|
|
+ val cid = rList(3)
|
|
|
+ val score = rList(2).replace("[", "").replace("]", "")
|
|
|
+ .split(",")(1).toDouble
|
|
|
+ val label = rList(0).toDouble
|
|
|
+ (cid, (1, label, score))
|
|
|
+ }).reduceByKey{
|
|
|
+ case (a, b) => (a._1 + b._1, a._2 + b._2, a._3 + b._3)
|
|
|
+ }.map{
|
|
|
+ case (cid, (all, zheng, scores)) =>
|
|
|
+ (cid, all, zheng, scores, zheng / all, scores / all)
|
|
|
+ }.collect().sortBy(_._1).map(_.productIterator.mkString("\t")).foreach(println)
|
|
|
+
|
|
|
}
|
|
|
|
|
|
|
|
|
- def createData(data: RDD[String], features: Array[String]): RDD[Row] = {
|
|
|
+ def createData4Ad(data: RDD[String], features: Array[String]): RDD[Row] = {
|
|
|
data.map(r => {
|
|
|
- val line: Array[String] = StringUtils.split(r, '\t')
|
|
|
+// val rList = r.split("\t")
|
|
|
+// val label = rList(0).toInt
|
|
|
+// val featureMap = scala.collection.mutable.Map[String, Double]()
|
|
|
+// var cid = -1
|
|
|
+// rList.drop(1).foreach(kv =>{
|
|
|
+// val kv_ = kv.split(":")
|
|
|
+// if (kv_(0).startsWith("cid_")){
|
|
|
+// cid = kv_(0).split("_")(1).toInt
|
|
|
+// }else{
|
|
|
+// featureMap.put(kv_(0), kv_(1).toDouble)
|
|
|
+// }
|
|
|
+// })
|
|
|
+// val v: Array[Any] = new Array[Any](features.length + 1)
|
|
|
+// v(0) = label
|
|
|
+//// v(1) = cid
|
|
|
+// for (i <- 0 until features.length) {
|
|
|
+// v(i + 1) = featureMap.getOrElse(r, 0.0D)
|
|
|
+// }
|
|
|
+// Row(v: _*)
|
|
|
+val line: Array[String] = StringUtils.split(r, '\t')
|
|
|
val label: Int = NumberUtils.toInt(line(0))
|
|
|
val map: util.Map[String, Double] = new util.HashMap[String, Double]
|
|
|
+ var cid = "-1"
|
|
|
for (i <- 1 until line.length) {
|
|
|
val fv: Array[String] = StringUtils.split(line(i), ':')
|
|
|
map.put(fv(0), NumberUtils.toDouble(fv(1), 0.0))
|
|
|
+ if(fv(0).startsWith("cid_")){
|
|
|
+ cid = fv(0).split("_")(1)
|
|
|
+ }
|
|
|
}
|
|
|
|
|
|
- val v: Array[Any] = new Array[Any](features.length + 1)
|
|
|
+ val v: Array[Any] = new Array[Any](features.length + 2)
|
|
|
v(0) = label
|
|
|
for (i <- 0 until features.length) {
|
|
|
v(i + 1) = map.getOrDefault(features(i), 0.0d)
|
|
|
}
|
|
|
+ v(features.length + 1) = cid
|
|
|
Row(v: _*)
|
|
|
})
|
|
|
}
|
