package com.imooc.bigdata.cp08.`trait`
import org.apache.spark.sql.SparkSession
//顶层数据处理接口trait DataProcess {
  def process(spark:SparkSession)
}

package com.imooc.bigdata.cp08.business
import com.imooc.bigdata.cp08.`trait`.DataProcessimport com.imooc.bigdata.cp08.utils.{IPUtils, KuduUtils, SQLUtils, SchemaUtils}import org.apache.spark.sql.SparkSession
object LogETLProcessor extends DataProcess{  override def process(spark: SparkSession): Unit = {
    //使用DataSourceAPI直接加载json数据    var jsonDF = spark.read.json("D:\\Hadoop基础与电商行为日志分析\\spark\\coding385\\sparksql-train\\data\\data-test.json")    //jsonDF.printSchema()    //jsonDF.show(false)
    //导入隐式转换    import spark.implicits._    //加载IP库,建议将RDD转成DF    val ipRowRDD = spark.sparkContext.textFile("D:\\Hadoop基础与电商行为日志分析\\spark\\coding385\\sparksql-train\\data\\ip.txt")    val ipRuleDF = ipRowRDD.map(x => {      val splits = x.split("\\|")      val startIP = splits(2).toLong      val endIP = splits(3).toLong      val province = splits(6)      val city = splits(7)      val isp = splits(9)
      (startIP, endIP, province, city, isp)    }).toDF("start_ip", "end_ip", "province", "city", "isp")    //ipRuleDF.show(false)
    //利用Spark SQL UDF转换json中的ip    import org.apache.spark.sql.functions._    def getLongIp() = udf((ip:String)=>{      IPUtils.ip2Long(ip)    })
    //添加字段传入十进制IP    jsonDF = jsonDF.withColumn("ip_long",      getLongIp()($"ip"))
    //将日志每一行的ip对应省份、城市、运行商进行解析    //两个DF进行join，条件是：json中的ip在规则ip中的范围内    //    val result = jsonDF.join(ipRuleDF, jsonDF("ip_long")    //      .between(ipRuleDF("start_ip"), ipRuleDF("end_ip")))    //      //.show(false)
    //用SQL的方式完成    jsonDF.createOrReplaceTempView("logs")    ipRuleDF.createOrReplaceTempView("ips")    val sql = SQLUtils.SQL    val result = spark.sql(sql)    //.show(false)
    //Kudu    val masterAddresses = "hadoop000"    val tableName = "ods"    val partitionId = "ip"    val schema = SchemaUtils.ODSSchema
    KuduUtils.sink(result,tableName,masterAddresses,schema,partitionId)    spark.read.format("org.apache.kudu.spark.kudu")      .option("kudu.master",masterAddresses)      .option("kudu.table",tableName)      .load().show()
  }
}

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 ProvinceCityStatProcessor extends DataProcess{  override def process(spark: SparkSession): Unit = {
    //从Kudu的ods表中读取数据，然后按照省份和城市分组即可    val sourceTableName = "ods"    val masterAddress = "hadoop000"
    val odsDF = spark.read.format("org.apache.kudu.spark.kudu")      .option("kudu.table", sourceTableName)      .option("kudu.master", masterAddress)      .load()    //odsDF.show(false)
    odsDF.createOrReplaceTempView("ods")    val result = spark.sql(SQLUtils.PROVINCE_CITY_SQL)    //result.show(false)
    //Kudu    val sinkTableName = "province_city_stat"    val partitionId = "provincename"    val schema = SchemaUtils.ProvinceCitySchema

    KuduUtils.sink(result,sinkTableName,masterAddress,schema,partitionId)    spark.read.format("org.apache.kudu.spark.kudu")      .option("kudu.master",masterAddress)      .option("kudu.table",sinkTableName)      .load().show()  }}

package com.imooc.bigdata.cp08
import com.imooc.bigdata.cp08.business.{LogETLProcessor, ProvinceCityStatProcessor}import org.apache.spark.sql.SparkSession
//整个项目的入口object SparkApp {
  def main(args: Array[String]): Unit = {    val spark = SparkSession.builder()      .master("local[2]")      .appName("SparkApp")      .getOrCreate()
    //ETL    LogETLProcessor.process(spark)    //省份    ProvinceCityStatProcessor.process(spark)
    spark.stop()  }
}