package com.imooc.bigdata.cp08.business import com.imooc.bigdata.cp08.`trait`.DataProcessimport com.imooc.bigdata.cp08.utils.{KuduUtils, SQLUtils, SchemaUtils}import org.apache.spark.sql.SparkSession object AreaStatProcessor extends DataProcess{  override def process(spark: SparkSession): Unit = {    val sourceTableName = "ods"    val masterAddresses = "hadoop000"     val odsDF = spark.read.format("org.apache.kudu.spark.kudu")      .option("kudu.table",sourceTableName)      .option("kudu.master",masterAddresses)      .load()     odsDF.createOrReplaceTempView("ods")     val resultTmp = spark.sql(SQLUtils.AREA_SQL_STEP1)    resultTmp.show()   }}

 lazy val AREA_SQL_STEP1 = "select provincename,cityname, " +    "sum(case when requestmode=1 and processnode >=1 then 1 else 0 end) origin_request," +    "sum(case when requestmode=1 and processnode >=2 then 1 else 0 end) valid_request," +    "sum(case when requestmode=1 and processnode =3 then 1 else 0 end) ad_request," +    "sum(case when adplatformproviderid>=100000 and iseffective=1 and isbilling=1 and isbid=1 and adorderid!=0 then 1 else 0 end) bid_cnt," +    "sum(case when adplatformproviderid>=100000 and iseffective=1 and isbilling=1 and iswin=1 then 1 else 0 end) bid_success_cnt," +    "sum(case when requestmode=2 and iseffective=1 then 1 else 0 end) ad_display_cnt," +    "sum(case when requestmode=3 and processnode=1 then 1 else 0 end) ad_click_cnt," +    "sum(case when requestmode=2 and iseffective=1 and isbilling=1 then 1 else 0 end) medium_display_cnt," +    "sum(case when requestmode=3 and iseffective=1 and isbilling=1 then 1 else 0 end) medium_click_cnt," +    "sum(case when adplatformproviderid>=100000 and iseffective=1 and isbilling=1 and iswin=1 and adorderid>20000  then 1*winprice/1000 else 0 end) ad_consumption," +    "sum(case when adplatformproviderid>=100000 and iseffective=1 and isbilling=1 and iswin=1 and adorderid>20000  then 1*adpayment/1000 else 0 end) ad_cost " +    "from ods group by provincename,cityname"

 AreaStatProcessor.process(spark)

  lazy val AREA_SQL_STEP2 = "select provincename,cityname, " +    "origin_request," +    "valid_request," +    "ad_request," +    "bid_cnt," +    "bid_success_cnt," +    "bid_success_cnt/bid_cnt bid_success_rate," +    "ad_display_cnt," +    "ad_click_cnt," +    "ad_click_cnt/ad_display_cnt ad_click_rate," +    "ad_consumption," +    "ad_cost from area_tmp " +    "where bid_cnt!=0 and ad_display_cnt!=0"

    resultTmp.createOrReplaceTempView("area_tmp")    val result = spark.sql(SQLUtils.AREA_SQL_STEP2)    result.show()

  lazy val AREASchema: Schema = {    val columns = List(      new ColumnSchemaBuilder("provincename",Type.STRING).nullable(false).key(true).build(),      new ColumnSchemaBuilder("cityname",Type.STRING).nullable(false).key(true).build(),      new ColumnSchemaBuilder("origin_request",Type.INT64).nullable(false).build(),      new ColumnSchemaBuilder("valid_request",Type.INT64).nullable(false).build(),      new ColumnSchemaBuilder("ad_request",Type.INT64).nullable(false).build(),      new ColumnSchemaBuilder("bid_cnt",Type.INT64).nullable(false).build(),      new ColumnSchemaBuilder("bid_success_cnt",Type.INT64).nullable(false).build(),      new ColumnSchemaBuilder("bid_success_rate",Type.DOUBLE).nullable(false).build(),      new ColumnSchemaBuilder("ad_display_cnt",Type.INT64).nullable(false).build(),      new ColumnSchemaBuilder("ad_click_cnt",Type.INT64).nullable(false).build(),      new ColumnSchemaBuilder("ad_click_rate",Type.DOUBLE).nullable(false).build(),      new ColumnSchemaBuilder("ad_consumption",Type.DOUBLE).nullable(false).build(),      new ColumnSchemaBuilder("ad_cost",Type.DOUBLE).nullable(false).build()    ).asJava

    val sinkTableName = "area_stat"    val partitionId = "provincename"    val schema = SchemaUtils.AREASchema     KuduUtils.sink(result,sinkTableName,masterAddresses,schema,partitionId)    spark.read.format("org.apache.kudu.spark.kudu")      .option("kudu.master",masterAddresses)      .option("kudu.table",sinkTableName)      .load().show()