1.x版本

1.x 版本使用SparkContext

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.hive.HiveContext
val conf = new SparkConf().set("spark.hadoop.validateOutputSpecs", "false").setAppName("mlx_feature_process")
val sc = new SparkContext(conf)
val predictSample = sc.textFile(...).map(r=>{PredictResult(r)}) 
val sqlCtx = new HiveContext(sc)
import sqlCtx.implicits._
predictSample.toDF().registerTempTable("predict_table")
val  tableName = "predict_sample_result_"+model
sqlCtx.sql("set hive.exec.dynamic.partition=true")
sqlCtx.sql("set hive.exec.dynamic.partition.mode=nostrick")
sqlCtx.sql("use ba_dealrank")
sqlCtx.sql(createTableSql(tableName))
sqlCtx.sql(
      s"""
         |alter table $tableName drop if exists partition(dt=$date)
         |""".stripMargin)
sqlCtx.sql(
      s"""
         |insert into $tableName partition(dt=$date)
         |select globalid,strategy,userid,item_id,item_type,
         |pctr,pcvr,pay_label,click_label, ts, hour, minute
         |from predict_table""".stripMargin)