update data processing

Author: delphi-xk

src/main/resources/param_tuning.md

@@ -7,12 +7,26 @@
 
 ## xgboost
 
-|object|booster|eval_metric|max_depth|eta|gamma|alpha|subsample|num_round|**train_result**|**valid_result**|**test_result**|
-|---|---|---|---|---|---|----|---|----|---|---|---|
-|reg:linear|gbtree|rmse|4|0.01|0.5|0.007|0.7|800|3.99 ~ 3.81|3.740|3.884|
-|reg:linear|gbtree|rmse|5|0.01|0.3|0.001|0.8|800|3.99 ~ 3.76|3.742|na|
-|reg:linear|gbtree|rmse|5|0.01|0.5|na|0.5|800|3.99 ~ 3.76|3.74|3.885|
-|reg:linear|gbtree|rmse|5|0.1|0.5|0.007|0.7|800|3.99 ~ 3.40|3.81|3.952|
-|reg:linear|gbtree|rmse|5|0.1|0.5|0.01|0.5|400|3.97 ~ 3.57|3.789|na|
-|reg:linear|gbtree|rmse|7|0.01|0.5|na|0.7|800|3.91 ~ 3.50|3.87|3.886|
-|reg:linear|gbtree|rmse|6|0.01|0.3|na|0.8|800|3.99 ~ 3.69|3.745|na|
+### object=reg:linear, booster=gbtree, eval_metric=rmse
+
+|max_depth|eta|gamma|alpha|subsample|min_child_weight|num_round|**train_result**|**valid_result**|**test_result**|
+|---|---|---|----|---|----|---|---|---|---|
+|4|0.01|0.5|0.007|0.7||800|3.99 ~ 3.81|3.740|3.884|
+|4|0.01|0.1| |0.7| 0.5|400|3.99 ~ 3.83|3.742|3.884|
+|4|0.005| 0.5|na|0.7|1|400|3.99 ~ 3.85|3.749|3.886|
+|5|0.01|0.5|na|0.5||800|3.99 ~ 3.76|3.74|3.885|
+|5|0.1|0.5|0.007|0.7||800|3.99 ~ 3.40|3.81|3.952|
+|5|0.1|0.5|0.01|0.5||400|3.97 ~ 3.57|3.789|na|
+|5|0.1|0.7|0.01|0.7|0.8|400|3.97 ~ 3.57|3.788|3.937|
+|5|0.01|0.7|0.01|0.7|0.8|400|3.99 ~ 3.80|3.741|3.884|
+|5|0.03|0.5|0.05|0.7| |400|3.99 ~ 3.732 |3.746 | |
+|5|0.03|0.5|0.03|0.7| |500| | | |
+|5|0.1|1|0.05|0.5|1|400|3.97 ~ 3.57|3.78|na|
+|5|0.5|1|0.05|0.5|1|400|3.89 ~ 3.30|4.19|na|
+|5|0.05|1|0.01|0.7|1|400|3.98 ~ 3.67 |3.754 |na|
+|5|0.03|1.5|0.05|0.7|1.5|400| 3.99 ~ 3.73|3.745|3.889|
+|6|0.01| na | 0.3|na |0.8|800|3.99 ~ 3.69|3.745|na|
+|6|0.02| 1.5 | 0.1|0.7|1|400|3.99 ~ 3.69|3.745 |na|
+|6|0.01| 1.0 | 0.1|0.7|1|400|3.99 ~ 3.76 |3.742 |na|
+|7|0.01|0.5|na|0.7||800|3.91 ~ 3.50|3.87|3.886|
+

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

@@ -205,16 +205,17 @@ object ModelPrediction {
  } else if (goal == "Regression"){
  val xgbParam = Map(
  "max_depth" -> 5,
- "alpha" -> 0.01f,
- "subsample" -> 0.5,
+ "alpha" -> 0.03f,
+ "subsample" -> 0.7,
  //"colsample_bytree" -> 0.7,
+ //"min_child_weight" -> 0.5,
  "objective" -> "reg:linear",
  //"top_k" -> "13",
  "booster" -> "gbtree",
- "eta" -> 0.1f,
+ "eta" -> 0.03f,
  "gamma" -> 0.5,
  "eval_metric" -> "rmse",
- "num_round" -> 400)
+ "num_round" -> 500)
  val xgbReg = new XGBoostRegressor(xgbParam)
  .setFeaturesCol("features").setLabelCol("label")
 

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

@@ -131,7 +131,8 @@ object NewDataProcess {
  }
 
  def getColumnAgg(transData:Dataset[Row], keyColumn:String, aggColumn:String): Dataset[Row] = {
- transData.groupBy(keyColumn)
+ transData.select(col(keyColumn), col(aggColumn).cast("double"))
+ .groupBy(keyColumn)
  .agg(sum(aggColumn).as(s"sum_$aggColumn"),
  avg(aggColumn).as(s"avg_$aggColumn")
  )
@@ -140,32 +141,76 @@ object NewDataProcess {
  def transProcess(): Unit = {
  val keyColumn = "card_id"
  val newTrans = spark.table("merchant.new_merchant_transactions")
- var ids = newTrans.select(keyColumn).distinct()
+ val ids = newTrans.select(keyColumn).distinct()
+ ids.cache()
+ val dropCols = Array("merchant_category_id", "subsector_id", "category_1", "city_id", "state_id", "category_2")
+ val merchants = spark.table("merchant.merchants").dropDuplicates(Array("merchant_id"))
+ .drop(dropCols: _*)
+ val tmpTrans = newTrans.join(merchants, Seq("merchant_id"), "left")
+ tmpTrans.cache()
+
+ var tmpModeRes = ids
  val processMode = Seq("city_id", "category_1", "installments", "category_3",
- "merchant_category_id", "category_2", "state_id", "subsector_id")
+ "merchant_category_id", "category_2", "state_id", "subsector_id",
+ "merchant_group_id", "category_4")
  for(colName <- processMode){
- val modeTmpTable = getColumnMode(newTrans, keyColumn, colName)
- ids = ids.join(modeTmpTable, Seq(keyColumn), "left")
+ val modeTmpTable = getColumnMode(tmpTrans, keyColumn, colName)
+ tmpModeRes = tmpModeRes.join(modeTmpTable, Seq(keyColumn), "left")
  }
- val aggTable = getColumnAgg(newTrans, keyColumn, "purchase_amount")
- ids = ids.join(aggTable, Seq(keyColumn), "left")
- saveTable(ids, "new_transactions_processed", "merchant")
+ saveTable(tmpModeRes, "tmp_mode_res", "merchant")
+
+ var tmpAggRes = ids
+ val aggCols = Array("purchase_amount", "numerical_1", "numerical_2", "avg_sales_lag3", "avg_purchases_lag3",
+ "avg_sales_lag6", "avg_purchases_lag6", "avg_sales_lag12", "avg_purchases_lag12")
+ for(aggCol <- aggCols){
+ val aggTable = getColumnAgg(tmpTrans, keyColumn, aggCol)
+ tmpAggRes = tmpAggRes.join(aggTable, Seq(keyColumn), "left")
+ }
+ saveTable(tmpAggRes, "tmp_agg_res", "merchant")
+
+ val modeRes = spark.table("merchant.tmp_mode_res")
+ val aggRes = spark.table("merchant.tmp_agg_res")
+ val tmpRes = modeRes.join(aggRes, Seq(keyColumn), "left")
+
+ saveTable(tmpRes, "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 hisTrans = spark.table("merchant.historical_transactions")
+ val ids = hisTrans.select(keyColumn).distinct()
+ ids.cache()
+ val dropCols = Array("merchant_category_id", "subsector_id", "category_1", "city_id", "state_id", "category_2")
+ val merchants = spark.table("merchant.merchants").dropDuplicates(Array("merchant_id"))
+ .drop(dropCols: _*)
+ val tmpTrans = hisTrans.join(merchants, Seq("merchant_id"), "left")
+ //tmpTrans.cache()
+
+ var tmpModeRes = ids
  val processMode = Seq("city_id", "category_1", "installments", "category_3",
- "merchant_category_id", "category_2", "state_id", "subsector_id")
+ "merchant_category_id", "category_2", "state_id", "subsector_id",
+ "merchant_group_id", "category_4")
  for(colName <- processMode){
- val modeTmpTable = getColumnMode(trans, keyColumn, colName)
- ids = ids.join(modeTmpTable, Seq(keyColumn), "left")
+ val modeTmpTable = getColumnMode(tmpTrans, keyColumn, colName)
+ tmpModeRes = tmpModeRes.join(modeTmpTable, Seq(keyColumn), "left")
  }
- val aggTable = getColumnAgg(trans, keyColumn, "purchase_amount")
- ids = ids.join(aggTable, Seq(keyColumn), "left")
- ids = addColumnsPrefix(ids, "historical", Array(keyColumn))
- saveTable(ids, "historical_transactions_processed", "merchant")
+ saveTable(tmpModeRes, "tmp_mode_res", "merchant")
+
+ var tmpAggRes = ids
+ val aggCols = Array("purchase_amount", "numerical_1", "numerical_2")
+ //"avg_sales_lag3", "avg_purchases_lag3", "avg_sales_lag6", "avg_purchases_lag6", "avg_sales_lag12", "avg_purchases_lag12")
+ for(aggCol <- aggCols){
+ val aggTable = getColumnAgg(tmpTrans, keyColumn, aggCol)
+ tmpAggRes = tmpAggRes.join(aggTable, Seq(keyColumn), "left")
+ }
+ saveTable(tmpAggRes, "tmp_agg_res", "merchant")
+
+ val modeRes = spark.table("merchant.tmp_mode_res")
+ val aggRes = spark.table("merchant.tmp_agg_res")
+ val tmpRes = addColumnsPrefix(modeRes.join(aggRes, Seq(keyColumn), "left"),
+ "historical", Array(keyColumn))
+
+ saveTable(tmpRes, "historical_transactions_processed", "merchant")
  }
 
  def merchantProcess(): Unit = {
@@ -186,8 +231,8 @@ object NewDataProcess {
  }
 
  def main(args: Array[String]): Unit = {
- transProcess()
- hisProcess()
+ //transProcess()
+ //hisProcess()
  merchantProcess()
  }