update model prediction

Author: delphi-xk

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

@@ -164,6 +164,7 @@ object NewDataProcess {
  }
  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")
  }
 
@@ -174,15 +175,19 @@ object NewDataProcess {
  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")
+ val hisTable = spark.table("merchant.historical_transactions_processed")
 
  trainTable = trainTable.join(transTable, Seq(keyColumn), "left")
+ .join(hisTable, Seq(keyColumn), "left")
  saveTable(trainTable, "train_result", "merchant")
  testTable = testTable.join(transTable, Seq(keyColumn), "left")
+ .join(hisTable, Seq(keyColumn), "left")
  saveTable(testTable, "test_result", "merchant")
  }
 
  def main(args: Array[String]): Unit = {
  transProcess()
+ hisProcess()
  merchantProcess()
  }
 

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

@@ -15,7 +15,7 @@ import org.apache.spark.sql.{Dataset, Row, SQLContext, SparkSession}
 import org.apache.spark.{SparkConf, SparkContext}
 import org.apache.spark.sql.types.{StringType, StructField, StructType}
 import org.apache.spark.util.SizeEstimator
-
+import org.apache.spark.sql.functions._
 import scala.util.Try
 
 
@@ -206,14 +206,14 @@ object SparkUtils {
  SizeEstimator.estimate(rdd)
  }
 
- // for test
-/* case class Person(name : String , age : Int)
-
- def createDatasetTest(): Unit ={
- val personRDD = sc.makeRDD(Seq(Person("A",10),Person("B",20)))
- val personDF = spark.createDataFrame(personRDD)
- val ds:Dataset[Person] = personDF.as[Person]
- }*/
+ def addColumnsPrefix(dataSet:Dataset[Row],
+ colPrefix:String,
+ ignoreCols:Array[String]): Dataset[Row] = {
+  dataSet.select(
+  dataSet.columns.map( fieldName =>
+  if(ignoreCols.contains(fieldName)) col(fieldName)
+  else col(fieldName).as(s"${colPrefix}__$fieldName") ): _*)
+ }
 
 }