update model prediction

Author: delphi-xk

src/main/scala/com/hyzs/spark/ml/ConvertLibsvm.scala

@@ -26,6 +26,7 @@ object ConvertLibsvm {
  val originalKey = "user_id"
  //val key = "user_id_md5"
  val key = "card_id"
+ val label = "target"
  val maxLabelMapLength = 100
  val convertPath = "/user/hyzs/convert/"
 
@@ -224,25 +225,26 @@ object ConvertLibsvm {
  val stringSchema = dataSchema.filter(field => field.dataType == StringType)
  val stringCols = stringSchema.map(field => field.name)
  val indexerArray = stringCols.map(field => getIndexers(dataSet, field))
- val objectArray = buildObjectArray(Array(key, "target"), dataSchema, indexerArray)
+ val objectArray = buildObjectArray(Array(key, label), dataSchema, indexerArray)
  val objRdd:RDD[String] = buildObjectJsonRdd(objectArray)
  val nameRdd = sc.makeRDD[String](dataSet.columns)
 
  println(s"start save obj: ${objRdd.first()} ...")
  saveRdd(objRdd, s"$taskPath$tableName.obj")
  println(s"start save name: ${nameRdd.first()} ...")
  saveRdd(nameRdd, s"$taskPath$tableName.name")
- println(s"merge obj files to $taskPath$tableName.obj.")
  objectArray
  }
 
  // dataSet id, target, feature1, feature2, ...
- def convertLibsvmFromDataSet(dataSet:Dataset[Row]): Unit = {
- val tableName = "tmpResult"
+ def convertLibsvmFromDataSet(dataSet:Dataset[Row], tableName:String, objs:Array[ModelObject]=null): Unit = {
  val taskPath = s"$convertPath$tableName/"
-
- val sourceData = processNull(dataSet)
- val objectArray = trainObjectArray(sourceData, tableName)
+ val processedData = processZeroValue(dataSet)
+ val sourceData = processNull(processedData)
+ var objectArray:Array[ModelObject] = null
+ if(objs == null){
+ objectArray = trainObjectArray(sourceData, tableName)
+ } else objectArray = objs
  val libsvm_result = replaceOldCols(sourceData, objectArray)
  val indexRdd:RDD[String] = libsvm_result.select(key).rdd.map(row => row(0).toString)
  val libsvmRdd: RDD[String] = libsvm_result.rdd.map(row => {
@@ -251,13 +253,19 @@ object ConvertLibsvm {
  })
  saveRdd(indexRdd, s"$taskPath$tableName.index")
  saveRdd(libsvmRdd, s"$taskPath$tableName.libsvm")
-
  }
 
- def main(args: Array[String]): Unit = {
- val data = spark.table("merchant.tmpResult")
- convertLibsvmFromDataSet(data)
 
+ def main(args: Array[String]): Unit = {
+ val trainName = "train_result"
+ val taskPath = s"$convertPath$trainName/"
+ val trainData = spark.table(s"merchant.$trainName")
+ convertLibsvmFromDataSet(trainData, trainName)
+
+ val testName = "test_result"
+ val testData =spark.table(s"merchant.$testName")
+ val objs = readObj(s"$taskPath$trainName.obj")
+ convertLibsvmFromDataSet(testData, testName, objs)
  }
 }
 

src/main/scala/com/hyzs/spark/ml/ModelPrediction.scala

@@ -1,5 +1,6 @@
 package com.hyzs.spark.ml
 
+import com.hyzs.spark.ml.ConvertLibsvm.saveRdd
 import com.hyzs.spark.mllib.evaluation.ConfusionMatrix
 import com.hyzs.spark.utils.SparkUtils._
 import org.apache.spark.ml.classification.GBTClassifier
@@ -9,10 +10,11 @@ import org.apache.spark.mllib.regression.LabeledPoint
 import org.apache.spark.mllib.tree.{GradientBoostedTrees, RandomForest}
 import org.apache.spark.rdd.RDD
 import org.apache.spark.mllib.tree.configuration.BoostingStrategy
-import org.apache.spark.mllib.tree.impurity.{Entropy, Gini}
+import org.apache.spark.mllib.tree.impurity.{Entropy, Gini, Variance}
 import org.apache.spark.mllib.util.MLUtils
 import ml.dmlc.xgboost4j.scala.spark.XGBoostClassifier
 import org.apache.spark.mllib.evaluation.BinaryClassificationMetrics
+import org.apache.spark.mllib.tree.model.GradientBoostedTreesModel
 /**
  * Created by xk on 2018/10/26.
  */
@@ -71,38 +73,59 @@ object ModelPrediction {
 
  def GBT(trainingData: RDD[LabeledPoint],
  validData: RDD[LabeledPoint],
- testData:RDD[LabeledPoint]): Unit = {
+ testData:RDD[LabeledPoint],
+ goal:String): Unit = {
+
 
  // Train a GradientBoostedTrees model.
  // The defaultParams for Classification use LogLoss by default.
- val boostingStrategy: BoostingStrategy = BoostingStrategy.defaultParams("Classification")
- boostingStrategy.setNumIterations(10) // Note: Use more iterations in practice. eg. 10, 20
- boostingStrategy.treeStrategy.setNumClasses(2)
- boostingStrategy.treeStrategy.setMaxDepth(6)
- // boostingStrategy.treeStrategy.setMaxBins(32)
+ // goal should be "Classification" or "Regression"
+ val boostingStrategy: BoostingStrategy = BoostingStrategy.defaultParams(goal)
+ boostingStrategy.setNumIterations(100) // Note: Use more iterations in practice. eg. 10, 20
+ boostingStrategy.setLearningRate(0.005)
+ //boostingStrategy.treeStrategy.setNumClasses(2)
+ boostingStrategy.treeStrategy.setMaxDepth(5)
+ boostingStrategy.treeStrategy.setImpurity(Variance)
+ boostingStrategy.treeStrategy.setMaxBins(32)
  // Empty categoricalFeaturesInfo indicates all features are continuous.
  // boostingStrategy.treeStrategy.categoricalFeaturesInfo = Map[Int, Int]()
  //boostingStrategy.treeStrategy.setImpurity(Entropy)
  //boostingStrategy.treeStrategy.setImpurity(Gini)
 
  // without validation
- // val model = GradientBoostedTrees.train(trainingData, boostingStrategy)
- val model = new GradientBoostedTrees(boostingStrategy).runWithValidation(trainingData, validData)
-
- val predAndLabels = testData.map { point =>
- val prediction = model.predict(point.features)
- (prediction, point.label)
- }.collect()
- val confusion = new ConfusionMatrix(predAndLabels)
+ val model = GradientBoostedTrees.train(trainingData, boostingStrategy)
+ //val model = new GradientBoostedTrees(boostingStrategy).runWithValidation(trainingData, validData)
+
+ if(goal == "Classification"){
+ val predAndLabels = testData.map { point =>
+ val prediction = model.predict(point.features)
+ (prediction, point.label)
+ }.collect()
+ val confusion = new ConfusionMatrix(predAndLabels)
+ println("model precision: " + confusion.precision)
+ println("model recall: " + confusion.recall)
+ println("model accuracy: " + confusion.accuracy)
+ println("model f1: " + confusion.f1_score)
+ } else if (goal == "Regression"){
+ val labelsAndPredictions = testData.map { point =>
+ val prediction = model.predict(point.features)
+ (point.label, prediction)
+ }
+ val testMSE = labelsAndPredictions.map{ case (v, p) => math.pow(v - p, 2) }.mean()
+ val rmse = math.sqrt(testMSE)
+ println(s"Test Mean Squared Error = $testMSE")
+ println(s"Root Mean Squared Error = $rmse")
+ println(s"Learned regression tree model:\n ${model.toDebugString}")
+ val modelPath = "/user/hyzs/model/gbt_regression"
+ println(s"save model to $modelPath")
+ if(checkHDFileExist(modelPath)) dropHDFiles(modelPath)
+ model.save(sc, modelPath)
+ } else throw new IllegalArgumentException(s"$goal is not supported by boosting.")
 
- println("model precision: " + confusion.precision)
- println("model recall: " + confusion.recall)
- println("model accuracy: " + confusion.accuracy)
- println("model f1: " + confusion.f1_score)
  }
 
 
- def GBT_ml(): Unit = {
+ def GBT_classifier(): Unit = {
  val data = spark.read.format("libsvm").load(libsvmPath)
  val Array(trainingData, testData) = data.randomSplit(Array(0.6, 0.4))
 
@@ -140,7 +163,6 @@ object ModelPrediction {
 
  }
 
-
  def xgboost_ml(): Unit = {
  val data = spark.read.format("libsvm").load(libsvmPath)
  val Array(trainingData, testData) = data.randomSplit(Array(0.6, 0.4))
@@ -168,15 +190,31 @@ object ModelPrediction {
 
  }
 
+ def predictModel(): Unit = {
+ val modelPath = "/user/hyzs/model/gbt_regression"
+ val dataPath = "/user/hyzs/convert/test_result/test_result.libsvm"
+ val model = GradientBoostedTreesModel.load(sc, modelPath)
+ val testData = MLUtils.loadLibSVMFile(sc, dataPath)
+ val preds = testData.map{ record =>
+ model.predict(record.features).toString
+ }
+ saveRdd(preds, "/user/hyzs/convert/test_result/preds.txt")
+
+ }
 
  def main(args: Array[String]): Unit = {
  /* val (trainingData, validData, testData) = prepareData()
  println("random forest =======")
  val randomForestModel = randomForest(trainingData, validData, testData)
  println("gbt =======")
  val gbt = GBT(trainingData, validData, testData)*/
- println("xgboost =======")
- xgboost_ml()
+/* println("xgboost =======")
+ xgboost_ml()*/
+ val rawData = MLUtils
+ .loadLibSVMFile(sc, "/user/hyzs/convert/train_result/train_result.libsvm")
+ .randomSplit(Array(0.7, 0.3))
+ GBT(rawData(0), null, rawData(1), "Regression")
+ predictModel()
 
  }
 }

src/main/scala/com/hyzs/spark/sql/NewDataProcess.scala

@@ -137,52 +137,53 @@ object NewDataProcess {
  )
  }
 
- def merchantProcess(): Unit = {
+ def transProcess(): Unit = {
  val keyColumn = "card_id"
- var trainTable = spark.table("merchant.train")
  val newTrans = spark.table("merchant.new_merchant_transactions")
+ var ids = newTrans.select(keyColumn).distinct()
  val processMode = Seq("city_id", "category_1", "installments", "category_3",
  "merchant_category_id", "category_2", "state_id", "subsector_id")
  for(colName <- processMode){
  val modeTmpTable = getColumnMode(newTrans, keyColumn, colName)
- trainTable = trainTable.join(modeTmpTable, Seq(keyColumn), "left")
+ ids = ids.join(modeTmpTable, Seq(keyColumn), "left")
  }
- saveTable(trainTable, "train_result", "merchant")
+ val aggTable = getColumnAgg(newTrans, keyColumn, "purchase_amount")
+ ids = ids.join(aggTable, Seq(keyColumn), "left")
+ saveTable(ids, "new_transactions_processed", "merchant")
+ }
 
+ def hisProcess(): Unit = {
+ val keyColumn = "card_id"
+ val trans = spark.table("merchant.historical_transactions")
+ var ids = trans.select(keyColumn).distinct()
+ val processMode = Seq("city_id", "category_1", "installments", "category_3",
+ "merchant_category_id", "category_2", "state_id", "subsector_id")
+ for(colName <- processMode){
+ val modeTmpTable = getColumnMode(trans, keyColumn, colName)
+ ids = ids.join(modeTmpTable, Seq(keyColumn), "left")
+ }
+ val aggTable = getColumnAgg(trans, keyColumn, "purchase_amount")
+ ids = ids.join(aggTable, Seq(keyColumn), "left")
+ saveTable(ids, "historical_transactions_processed", "merchant")
+ }
 
+ def merchantProcess(): Unit = {
+ val keyColumn = "card_id"
+ var trainTable = spark.table("merchant.train")
+ .select("card_id", "target", "feature_1", "feature_2", "feature_3")
+ var testTable = spark.table("merchant.test").withColumn("target", lit(0))
+ .select("card_id", "target", "feature_1", "feature_2", "feature_3")
+ val transTable = spark.table("merchant.new_transactions_processed")
 
+ trainTable = trainTable.join(transTable, Seq(keyColumn), "left")
+ saveTable(trainTable, "train_result", "merchant")
+ testTable = testTable.join(transTable, Seq(keyColumn), "left")
+ saveTable(testTable, "test_result", "merchant")
  }
 
  def main(args: Array[String]): Unit = {
- //preprocessOrder()
- val key = "id"
- val table = spark.table("sample_n_enc")
- val sample1 = spark.table("sample_w_enc")
- val rowTable = columnToRow(table)
- saveTable(rowTable, "sample_2")
-
- val sample_features = sample1.join(table, Seq("id"), "left")
- saveTable(sample_features, "sample_features")
-
- val all = spark.table("hyzs.all_data")
-
- //val diffCols = all.columns diff sample_features.columns
-
- val order = spark.table("sample_order")
- //spark.sql("drop table sample_fix")
- //spark.sql("create table sample_fix(id string, brs_brs_p0001308 string, mkt_schd_p0001328 string, mkt_schd_p0001327 string)")
-
- val fix = spark.table("sample_fix")
- val features = sample_features.join(fix, Seq("id"), "left")
- .selectExpr("id"+:(all.columns diff Seq("user_id", "user_id_md5")): _*)
- saveTable(features, "features")
- val sample_all = order.join(features, Seq("id"), "right")
- saveTable(sample_all, "sample_all")
- sample_all
- .coalesce(1)
- .write.format("com.databricks.spark.csv")
- .option("header", "true")
- .save("/hyzs/test_data/sample_all.csv")
+ transProcess()
+ merchantProcess()
  }
 
 }

src/main/scala/com/hyzs/spark/utils/SparkUtils.scala

@@ -36,7 +36,7 @@ object SparkUtils {
  val hdConf: Configuration = sc.hadoopConfiguration
  val fs: FileSystem = FileSystem.get(hdConf)
 
- val warehouseDir: String = conf.getOption("spark.sql.warehouse.dir").getOrElse("/user/hive/warehouse/")
+ val warehouseDir: String = conf.getOption("spark.sql.warehouse.dir").getOrElse("/user/hive/warehouse")
  val partitionNums: Int = conf.getOption("spark.sql.shuffle.partitions").getOrElse("200").toInt
  val invalidRowPath = "/hyzs/invalidRows/"
  val mapper = new ObjectMapper().configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
@@ -110,21 +110,28 @@ object SparkUtils {
  def processNull(df: Dataset[Row]): Dataset[Row] = {
  df.na.fill(0.0)
  .na.fill("0.0")
- .na.replace("*", Map("" -> "0.0", "null" -> "0.0", -9999 -> 0.0))
+ .na.replace("*", Map("" -> "0.0", "null" -> "0.0"))
  }
 
+ def processZeroValue(df: Dataset[Row]): Dataset[Row] = {
+ df.na
+ .replace("*", Map(0 -> 1, 0.0 -> 1))
+ }
+
+
  def saveTable(df: Dataset[Row], tableName:String, dbName:String = "default"): Unit = {
 
  spark.sql(s"drop table if exists $dbName.$tableName")
  var path = ""
  if(dbName != "default"){
- path = s"$warehouseDir$dbName.db/$tableName"
+ path = s"$warehouseDir/$dbName.db/$tableName"
  }
  else{
- path = s"$warehouseDir$tableName"
+ path = s"$warehouseDir/$tableName"
  }
  if(checkHDFileExist(path))dropHDFiles(path)
  df.write
+ .option("path", path)
  .saveAsTable(s"$dbName.$tableName")
  }