客快物流大数据项目(六十三)：快递单主题

SELECT EBILL."id" ,
EBILL ."express_number" ,
EBILL ."cid" ,
CUSTOMER."name" AS cname,
ADDRESS ."detail_addr" AS caddress,
EBILL."eid" ,
COURIER ."name" AS ename,
DOT ."id" AS dot_id,
DOT ."dot_name" AS dot_name,
COMPANY ."company_name" AS company_name,
EBILL."order_channel_id" ,
code1."name" AS "order_channel_name",
ebill."order_dt",
ebill."order_terminal_type",
code2."name" AS order_terminal_type_name,
ebill."order_terminal_os_type" ,
ebill."reserve_dt" ,
ebill."is_collect_package_timeout" ,
ebill."timeout_dt" ,
CUSTOMER."type" ,
ebill."cdt" ,
ebill."udt" ,
ebill."remark" 
FROM "tbl_express_bill" EBILL
LEFT JOIN "tbl_courier" courier ON EBILL."eid" = courier."id" 
LEFT JOIN "tbl_customer" customer ON ebill."cid" = CUSTOMER ."id" 
LEFT JOIN "tbl_codes" code1 ON code1."type" =18 AND ebill."order_channel_id" =code1."code"  
LEFT JOIN "tbl_codes" code2 ON code2."type" =17 AND ebill."order_terminal_type" =code2."code"  
LEFT JOIN "tbl_consumer_address_map" address_map ON CUSTOMER."id" = address_map."consumer_id"
LEFT JOIN "tbl_address" address ON address_map."address_id" = ADDRESS."id"
LEFT JOIN "tbl_pkg" pkg ON EBILL."express_number" = pkg."pw_bill"
LEFT JOIN "tbl_dot" dot ON PKG ."pw_dot_id"=dot."id" 
LEFT JOIN "tbl_company_dot_map" companydot ON dot."id" =COMPANYDOT ."dot_id" 
LEFT JOIN "tbl_company" company ON COMPANY ."id"=companydot."company_id" 

package cn.it.logistics.offline.dwd

import cn.it.logistics.common.{CodeTypeMapping, Configuration, OfflineTableDefine, SparkUtils}
import cn.it.logistics.offline.OfflineApp
import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.storage.StorageLevel
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.IntegerType

/**
 * 快递单主题开发
 * 将快递单事实表的数据与相关维度表的数据进行关联，然后将拉宽后的数据写入到快递单宽表中
 * 采用DSL语义实现离线计算程序
 *
 * 最终离线程序需要部署到服务器，每天定时执行（azkaban定时调度）
 */
object ExpressBillDWD extends OfflineApp {
  //定义应用的名称
  val appName = this.getClass.getSimpleName

/**
   * 入口函数
   * @param args
   */
  def main(args: Array[String]): Unit = {
    /**
     * 实现步骤：
     * 1）初始化sparkConf对象
     * 2）创建sparkSession对象
     * 3）加载kudu中的事实表和维度表的数据（将加载后的数据进行缓存）
     * 4）定义维度表与事实表的关联
     * 5）将拉宽后的数据再次写回到kudu数据库中（DWD明细层）
     * 5.1：创建快递单明细宽表的schema表结构
     * 5.2：创建快递单宽表（判断宽表是否存在，如果不存在则创建）
     * 5.3：将数据写入到kudu中
     * 6)：将缓存的数据删除掉
     */

    //1）初始化sparkConf对象
    val sparkConf: SparkConf = SparkUtils.autoSettingEnv(
      SparkUtils.sparkConf(appName)
    )

    //2）创建sparkSession对象
    val sparkSession: SparkSession = SparkUtils.getSparkSession(sparkConf)
    sparkSession.sparkContext.setLogLevel(Configuration.LOG_OFF)

    //数据处理
    execute(sparkSession)
  }

  /**
   * 数据处理
   *
   * @param sparkSession
   */
  override def execute(sparkSession: SparkSession): Unit = {
    sparkSession.stop()
  }
}

//3.1：加载快递单事实表的数据
val expressBillDF: DataFrame = getKuduSource(sparkSession, TableMapping.expressBill, Configuration.isFirstRunnable)
  .persist(StorageLevel.DISK_ONLY_2) //将数据缓存两个节点的磁盘目录，避免单机故障导致的缓存数据丢失

//3.2：加载快递员维度表的数据
val courierDF: DataFrame = getKuduSource(sparkSession, TableMapping.courier, true).persist(StorageLevel.DISK_ONLY_2)

//3.2：加载客户维度表的数据
val customerDF: DataFrame = getKuduSource(sparkSession, TableMapping.customer, true).persist(StorageLevel.DISK_ONLY_2)

//3.4：加载物流码表的数据
val codesDF: DataFrame = getKuduSource(sparkSession, TableMapping.codes, true).persist(StorageLevel.DISK_ONLY_2)

//3.5：客户地址关联表的数据
val addressMapDF: DataFrame = getKuduSource(sparkSession, TableMapping.consumerAddressMap, true).persist(StorageLevel.DISK_ONLY_2)

//3.6：加载地址表的数据
val addressDF: DataFrame = getKuduSource(sparkSession, TableMapping.address, true).persist(StorageLevel.DISK_ONLY_2)

//3.7：加载包裹表的数据
val pkgDF: DataFrame = getKuduSource(sparkSession, TableMapping.pkg, true).persist(StorageLevel.DISK_ONLY_2)

//3.8：加载网点表的数据
val dotDF: DataFrame = getKuduSource(sparkSession, TableMapping.dot, true).persist(StorageLevel.DISK_ONLY_2)

//3.9：加载公司网点表的数据
val companyDotMapDF: DataFrame = getKuduSource(sparkSession, TableMapping.companyDotMap, true).persist(StorageLevel.DISK_ONLY_2)

//3.10：加载公司表的数据
val companyDF: DataFrame = getKuduSource(sparkSession, TableMapping.company, true).persist(StorageLevel.DISK_ONLY_2)

//导入隐式转换
import  sparkSession.implicits._

//获取终端类型码表数据
val orderTerminalTypeDF: DataFrame = codesDF.where($"type" === CodeTypeMapping.OrderTerminalType)
  .select($"code".as("OrderTerminalTypeCode"), $"codeDesc".as("OrderTerminalTypeName"))

//获取下单终端类型码表数据
val orderChannelTypeDF: DataFrame = codesDF.where($"type" === CodeTypeMapping.OrderChannelType)
  .select($"code".as("OrderChannelTypeCode"), $"codeDesc".as("OrderChannelTypeName"))

//TODO 4）定义维度表与事实表的关联
val joinType = "left_outer"
val expressBillDetailDF: DataFrame = expressBillDF
  .join(courierDF, expressBillDF("eid") === courierDF("id") ,joinType) //快递单表与快递员表进行关联
  .join(customerDF, expressBillDF("cid")  === customerDF("id"), joinType) //快递单表与客户表进行关联
  .join(orderChannelTypeDF, orderChannelTypeDF("OrderChannelTypeCode") === expressBillDF("orderChannelId"), joinType) //下单渠道表与快递单表关联
  .join(orderTerminalTypeDF, orderTerminalTypeDF("OrderTerminalTypeCode") === expressBillDF("orderTerminalType"), joinType) //终端类型表与快递单表关联
  .join(addressMapDF, addressMapDF("consumerId") === customerDF("id"), joinType) //客户地址关联表与客户表关联
  .join(addressDF, addressDF("id") === addressMapDF("addressId"), joinType)  //地址表与客户地址关联表关联
  .join(pkgDF, pkgDF("pwBill") === expressBillDF("expressNumber"), joinType) //包裹表与快递单表关联
  .join(dotDF, dotDF("id") === pkgDF("pwDotId"), joinType)  //网点表与包裹表关联
  .join(companyDotMapDF, companyDotMapDF("dotId") === dotDF("id"), joinType) //公司网点关联表与网点表关联
  .join(companyDF, companyDF("id") === companyDotMapDF("companyId"), joinType) //公司网点关联表与公司表关联
  .withColumn("day", date_format(expressBillDF("cdt"), "yyyyMMdd"))//虚拟列,可以根据这个日期列作为分区字段，可以保证同一天的数据保存在同一个分区中
  .sort(expressBillDF.col("cdt").asc) //根据快递单的创建时间顺序排序
  .select(
    expressBillDF("id"),        //快递单id
    expressBillDF("expressNumber").as("express_number"),  //快递单编号
    expressBillDF("cid"),     //客户id
    customerDF("name").as("cname"), //客户名称
    addressDF("detailAddr").as("caddress"),      //客户地址
    expressBillDF("eid"), //员工id
    courierDF("name").as("ename"), //员工名称
    dotDF("id").as("dot_id"), //网点id
    dotDF("dotName").as("dot_name"), //网点名称
    companyDF("companyName").as("company_name"),//公司名称
    expressBillDF("orderChannelId").as("order_channel_id"), //下单渠道id
    orderChannelTypeDF("OrderChannelTypeName").as("order_channel_name"), //下单渠道id
    expressBillDF("orderDt").as("order_dt"),//下单时间
    orderTerminalTypeDF("OrderTerminalTypeCode").as("order_terminal_type"), //下单设备类型id
    orderTerminalTypeDF("OrderTerminalTypeName").as("order_terminal_type_name"), //下单设备类型id
    expressBillDF("orderTerminalOsType").as("order_terminal_os_type"),//下单设备操作系统
    expressBillDF("reserveDt").as("reserve_dt"),//预约取件时间
    expressBillDF("isCollectPackageTimeout").as("is_collect_package_timeout"),//是否取件超时
    expressBillDF("timeoutDt").as("timeout_dt"),//超时时间
    customerDF("type"),//客户类型
    expressBillDF("cdt"),//创建时间
    expressBillDF("udt"),//修改时间
    expressBillDF("remark"),//备注
    $"day"
  )

//TODO 5）将拉宽后的数据再次写回到kudu数据库中（DWD明细层）
save(expressBillDetailDF, OfflineTableDefine.expressBillDetail)

//TODO 6) 将缓存的数据删除掉
expressBillDF.unpersist()
courierDF.unpersist()
customerDF.unpersist()
orderChannelTypeDF.unpersist()
orderTerminalTypeDF.unpersist()
addressMapDF.unpersist()
addressDF.unpersist()
pkgDF.unpersist()
dotDF.unpersist()
companyDotMapDF.unpersist()
companyDF.unpersist()

package cn.it.logistics.offline.dwd

import cn.it.logistics.common.{CodeTypeMapping, Configuration, OfflineTableDefine, SparkUtils}
import cn.it.logistics.offline.OfflineApp
import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.storage.StorageLevel
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.IntegerType

/**
 * 快递单主题开发
 * 将快递单事实表的数据与相关维度表的数据进行关联，然后将拉宽后的数据写入到快递单宽表中
 * 采用DSL语义实现离线计算程序
 *
 * 最终离线程序需要部署到服务器，每天定时执行（azkaban定时调度）
 */
object ExpressBillDWD extends OfflineApp {
  //定义应用的名称
  val appName = this.getClass.getSimpleName

/**
   * 入口函数
   * @param args
   */
  def main(args: Array[String]): Unit = {
    /**
     * 实现步骤：
     * 1）初始化sparkConf对象
     * 2）创建sparkSession对象
     * 3）加载kudu中的事实表和维度表的数据（将加载后的数据进行缓存）
     * 4）定义维度表与事实表的关联
     * 5）将拉宽后的数据再次写回到kudu数据库中（DWD明细层）
     * 5.1：创建快递单明细宽表的schema表结构
     * 5.2：创建快递单宽表（判断宽表是否存在，如果不存在则创建）
     * 5.3：将数据写入到kudu中
     * 6)：将缓存的数据删除掉
     */

    //1）初始化sparkConf对象
    val sparkConf: SparkConf = SparkUtils.autoSettingEnv(
      SparkUtils.sparkConf(appName)
    )

    //2）创建sparkSession对象
    val sparkSession: SparkSession = SparkUtils.getSparkSession(sparkConf)
    sparkSession.sparkContext.setLogLevel(Configuration.LOG_OFF)

    //数据处理
    execute(sparkSession)
  }

  /**
   * 数据处理
   *
   * @param sparkSession
   */
  override def execute(sparkSession: SparkSession): Unit = {
    //TODO 3）加载kudu中的事实表和维度表的数据（将加载后的数据进行缓存）
    //3.1：加载快递单事实表的数据
    val expressBillDF: DataFrame = getKuduSource(sparkSession, "tbl_express_bill", Configuration.isFirstRunnable)
      .persist(StorageLevel.DISK_ONLY_2) //将数据缓存两个节点的磁盘目录，避免单机故障导致的缓存数据丢失

    //3.2：加载快递员维度表的数据
    val courierDF: DataFrame = getKuduSource(sparkSession, "tbl_courier", true).persist(StorageLevel.DISK_ONLY_2)

    //3.2：加载客户维度表的数据
    val customerDF: DataFrame = getKuduSource(sparkSession, "tbl_customer", true).persist(StorageLevel.DISK_ONLY_2)

    //3.4：加载物流码表的数据
    val codesDF: DataFrame = getKuduSource(sparkSession, "tbl_codes", true).persist(StorageLevel.DISK_ONLY_2)

    //3.5：客户地址关联表的数据
    val addressMapDF: DataFrame = getKuduSource(sparkSession, "tbl_consumer_address_map", true).persist(StorageLevel.DISK_ONLY_2)

    //3.6：加载地址表的数据
    val addressDF: DataFrame = getKuduSource(sparkSession, "tbl_address", true).persist(StorageLevel.DISK_ONLY_2)

    //3.7：加载包裹表的数据
    val pkgDF: DataFrame = getKuduSource(sparkSession, "tbl_pkg", true).persist(StorageLevel.DISK_ONLY_2)

    //3.8：加载网点表的数据
    val dotDF: DataFrame = getKuduSource(sparkSession, "tbl_dot", true).persist(StorageLevel.DISK_ONLY_2)

    //3.9：加载公司网点表的数据
    val companyDotMapDF: DataFrame = getKuduSource(sparkSession, "tbl_company_dot_map", true).persist(StorageLevel.DISK_ONLY_2)

    //3.10：加载公司表的数据
    val companyDF: DataFrame = getKuduSource(sparkSession, "tbl_company", true).persist(StorageLevel.DISK_ONLY_2)

    //导入隐式转换
    import  sparkSession.implicits._

    //获取终端类型码表数据
    val orderTerminalTypeDF: DataFrame = codesDF.where($"type" === CodeTypeMapping.OrderTerminalType)
      .select($"code".as("OrderTerminalTypeCode"), $"codeDesc".as("OrderTerminalTypeName"))

    //获取下单终端类型码表数据
    val orderChannelTypeDF: DataFrame = codesDF.where($"type" === CodeTypeMapping.OrderChannelType)
      .select($"code".as("OrderChannelTypeCode"), $"codeDesc".as("OrderChannelTypeName"))

    //TODO 4）定义维度表与事实表的关联
    val joinType = "left_outer"
    val expressBillDetailDF: DataFrame = expressBillDF
      .join(courierDF, expressBillDF("eid") === courierDF("id") ,joinType) //快递单表与快递员表进行关联
      .join(customerDF, expressBillDF("cid")  === customerDF("id"), joinType) //快递单表与客户表进行关联
      .join(orderChannelTypeDF, orderChannelTypeDF("OrderChannelTypeCode") === expressBillDF("orderChannelId"), joinType) //下单渠道表与快递单表关联
      .join(orderTerminalTypeDF, orderTerminalTypeDF("OrderTerminalTypeCode") === expressBillDF("orderTerminalType"), joinType) //终端类型表与快递单表关联
      .join(addressMapDF, addressMapDF("consumerId") === customerDF("id"), joinType) //客户地址关联表与客户表关联
      .join(addressDF, addressDF("id") === addressMapDF("addressId"), joinType)  //地址表与客户地址关联表关联
      .join(pkgDF, pkgDF("pwBill") === expressBillDF("expressNumber"), joinType) //包裹表与快递单表关联
      .join(dotDF, dotDF("id") === pkgDF("pwDotId"), joinType)  //网点表与包裹表关联
      .join(companyDotMapDF, companyDotMapDF("dotId") === dotDF("id"), joinType) //公司网点关联表与网点表关联
      .join(companyDF, companyDF("id") === companyDotMapDF("companyId"), joinType) //公司网点关联表与公司表关联
      .withColumn("day", date_format(expressBillDF("cdt"), "yyyyMMdd"))//虚拟列,可以根据这个日期列作为分区字段，可以保证同一天的数据保存在同一个分区中
      .sort(expressBillDF.col("cdt").asc) //根据快递单的创建时间顺序排序
      .select(
        expressBillDF("id"),        //快递单id
        expressBillDF("expressNumber").as("express_number"),  //快递单编号
        expressBillDF("cid"),     //客户id
        customerDF("name").as("cname"), //客户名称
        addressDF("detailAddr").as("caddress"),      //客户地址
        expressBillDF("eid"), //员工id
        courierDF("name").as("ename"), //员工名称
        dotDF("id").as("dot_id"), //网点id
        dotDF("dotName").as("dot_name"), //网点名称
        companyDF("companyName").as("company_name"),//公司名称
        expressBillDF("orderChannelId").as("order_channel_id"), //下单渠道id
        orderChannelTypeDF("OrderChannelTypeName").as("order_channel_name"), //下单渠道id
        expressBillDF("orderDt").as("order_dt"),//下单时间
        orderTerminalTypeDF("OrderTerminalTypeCode").as("order_terminal_type"), //下单设备类型id
        orderTerminalTypeDF("OrderTerminalTypeName").as("order_terminal_type_name"), //下单设备类型id
        expressBillDF("orderTerminalOsType").as("order_terminal_os_type"),//下单设备操作系统
        expressBillDF("reserveDt").as("reserve_dt"),//预约取件时间
        expressBillDF("isCollectPackageTimeout").as("is_collect_package_timeout"),//是否取件超时
        expressBillDF("timeoutDt").as("timeout_dt"),//超时时间
        customerDF("type"),//客户类型
        expressBillDF("cdt"),//创建时间
        expressBillDF("udt"),//修改时间
        expressBillDF("remark"),//备注
        $"day"
      )

    //TODO 5）将拉宽后的数据再次写回到kudu数据库中（DWD明细层）
    save(expressBillDetailDF, OfflineTableDefine.expressBillDetail)

    //TODO 6) 将缓存的数据删除掉
    expressBillDF.unpersist()
    courierDF.unpersist()
    customerDF.unpersist()
    orderChannelTypeDF.unpersist()
    orderTerminalTypeDF.unpersist()
    addressMapDF.unpersist()
    addressDF.unpersist()
    pkgDF.unpersist()
    dotDF.unpersist()
    companyDotMapDF.unpersist()
    companyDF.unpersist()

    sparkSession.stop()
  }
}

// 检查今日数据
spark.read
  .format(Configure.SPARK_KUDU_FORMAT)
  .options(Map("kudu.master" -> Configure.kuduRpcAddress, "kudu.table" -> table))
  .load
  .show

package cn.it.logistics.offline.dws

import cn.it.logistics.common.{Configuration, SparkUtils}
import cn.it.logistics.offline.OfflineApp
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession

object ExpressBillDWS  extends OfflineApp{
  //定义应用程序的名称
  val appName = this.getClass.getSimpleName

  /**
   * 入口函数
   * @param args
   */
  def main(args: Array[String]): Unit = {
    /**
     * 实现步骤：
     * 1）创建SparkConf对象
     * 2）创建SparkSession对象
     * 3）读取快递明细宽表的数据
     * 4）对快递明细宽表的数据进行指标的计算
     * 5）将计算好的指标数据写入到kudu数据库中
     * 5.1：定义指标结果表的schema信息
     * 5.2：组织需要写入到kudu表的数据
     * 5.3：判断指标结果表是否存在，如果不存在则创建
     * 5.4：将数据写入到kudu表中
     * 6）删除缓存数据
     * 7）停止任务，退出sparksession
     */

    //TODO 1）创建SparkConf对象
    val sparkConf: SparkConf = SparkUtils.autoSettingEnv(
      SparkUtils.sparkConf(appName)
    )

    //TODO 2）创建SparkSession对象
    val sparkSession: SparkSession = SparkUtils.getSparkSession(sparkConf)
    sparkSession.sparkContext.setLogLevel(Configuration.LOG_OFF)

    //处理数据
    execute(sparkSession)
  }

  /**
   * 数据处理
   *
   * @param sparkSession
   */
  override def execute(sparkSession: SparkSession): Unit = {
    // 退出sc
    sparkSession.stop
  }
}

//TODO 3）读取快递单明细宽表的数据
val expressBillDetailDF: DataFrame = getKuduSource(sparkSession, OfflineTableDefine.expressBillDetail, Configuration.isFirstRunnable)

//读取出来的明细宽表数据可能是增量的数据，也可能是全量的数据
//全量的数据是包含多个日期的数据，增量数据是前一天的数据
//需要计算的指标是以日为单位，每天的最大快递单数、最小快递单数、平均快递单数据
//因此需要对读取出来的快递单明细宽表数据按照日为单位进行分组，然后统计每日的指标数据
val expressBillDetailGroupByDayDF: DataFrame = expressBillDetailDF.select("day").groupBy("day").count().cache()

//导入隐式转换
import sparkSession.implicits._

//定义计算好的指标结果集合对象
val rows: ArrayBuffer[Row] = ArrayBuffer[Row]()

//循环遍历所有日期的数据
expressBillDetailGroupByDayDF.collect().foreach(row=>{
  //获取到需要处理的数据所在的日期
  val day: String = row.getAs[String](0)

  //根据日期查询该日期内的快递单明细数据，然后将查询到的结果进行指标计算（指定日期的指标）
  val expressBillDetailByDayDF: DataFrame = expressBillDetailDF.where(col("day") === day).toDF().persist(StorageLevel.DISK_ONLY_2)

  //TODO 4）对快递明细宽表的数据进行指标的计算
  //总快递单数
  val totalExpressBillCount: Long = expressBillDetailByDayDF.agg(count("express_number")).first().getLong(0)
  //各类客户的快递单数
  val customerTypeExpressBillTotalCountDF: DataFrame = expressBillDetailByDayDF.groupBy($"type").agg(count("express_number").as("express_number")).cache()

  //各类客户最大快递单数
  val maxTypeExpressBillTotalCount: Row = customerTypeExpressBillTotalCountDF.agg(max("express_number")).first()

  //各类客户最小快递单数
  val minTypeExpressBillTotalCount: Row = customerTypeExpressBillTotalCountDF.agg(min("express_number")).first()

  //各类客户平均快递单数
  val avgTypeExpressBillTotalCount: Row = customerTypeExpressBillTotalCountDF.agg(avg("express_number")).first()

  //各网点的快递单数
  val dotExpressBillTotalCountDF: DataFrame = expressBillDetailByDayDF.groupBy($"dot_id").agg(count("express_number").as("express_number")).cache()

  //各网点最大快递单数
  val maxDotExpressBillTotalCount: Row = dotExpressBillTotalCountDF.agg(max("express_number")).first()

  //各网点最小快递单数
  val minDotExpressBillTotalCount: Row = dotExpressBillTotalCountDF.agg(min("express_number")).first()

  //各网点平均快递单数
  val avgDotExpressBillTotalCount: Row = dotExpressBillTotalCountDF.agg(avg("express_number")).first()

  //各渠道的快递单数
  val channelExpressBillTotalCountDF: DataFrame = expressBillDetailByDayDF.groupBy($"order_channel_id").agg(count("express_number").as("express_number")).cache()

  //各渠道最大快递单数
  val maxChannelExpressBillTotalCount: Row = channelExpressBillTotalCountDF.agg(max("express_number")).first()

  //各渠道最小快递单数
  val minChannelExpressBillTotalCount: Row = channelExpressBillTotalCountDF.agg(min("express_number")).first()

  //各渠道平均快递单数
  val avgChannelExpressBillTotalCount: Row = channelExpressBillTotalCountDF.agg(avg("express_number")).first()

  val terminalExpressBillTotalCountDF: DataFrame = expressBillDetailByDayDF.groupBy($"order_terminal_type").agg(count("express_number").as("express_number")).cache()
  println(terminalExpressBillTotalCountDF)

  //各终端最大快递单数
  val maxTerminalExpressBillTotalCount: Row = terminalExpressBillTotalCountDF.agg(max("express_number")).first()

  //各终端最小快递单数
  val minTerminalExpressBillTotalCount: Row = terminalExpressBillTotalCountDF.agg(min("express_number")).first()

  //各终端平均快递单数
  val avgTerminalExpressBillTotalCount: Row = terminalExpressBillTotalCountDF.agg(avg("express_number")).first()
  println(avgTerminalExpressBillTotalCount)

  //将每条记录写入到Row对象中
  val rowInfo = Row(
    day,
    totalExpressBillCount, //总快递单数
    if(maxTypeExpressBillTotalCount.isNullAt(0)) 0L else maxTypeExpressBillTotalCount.get(0).asInstanceOf[Number].longValue(),
    if(minTypeExpressBillTotalCount.isNullAt(0)) 0L else minTypeExpressBillTotalCount.get(0).asInstanceOf[Number].longValue(),
    if(avgTypeExpressBillTotalCount.isNullAt(0)) 0L else avgTypeExpressBillTotalCount.get(0).asInstanceOf[Number].longValue(),
    if(maxDotExpressBillTotalCount.isNullAt(0)) 0L else maxDotExpressBillTotalCount.get(0).asInstanceOf[Number].longValue(),
    if(minDotExpressBillTotalCount.isNullAt(0)) 0L else minDotExpressBillTotalCount.get(0).asInstanceOf[Number].longValue(),
    if(avgDotExpressBillTotalCount.isNullAt(0)) 0L else avgDotExpressBillTotalCount.get(0).asInstanceOf[Number].longValue(),
    if(maxChannelExpressBillTotalCount.isNullAt(0)) 0L else maxChannelExpressBillTotalCount.get(0).asInstanceOf[Number].longValue(),
    if(minChannelExpressBillTotalCount.isNullAt(0)) 0L else minChannelExpressBillTotalCount.get(0).asInstanceOf[Number].longValue(),
    if(avgChannelExpressBillTotalCount.isNullAt(0)) 0L else avgChannelExpressBillTotalCount.get(0).asInstanceOf[Number].longValue(),
    if(maxTerminalExpressBillTotalCount.isNullAt(0)) 0L else maxTerminalExpressBillTotalCount.get(0).asInstanceOf[Number].longValue(),
    if(minTerminalExpressBillTotalCount.isNullAt(0)) 0L else minTerminalExpressBillTotalCount.get(0).asInstanceOf[Number].longValue(),
    if(avgTerminalExpressBillTotalCount.isNullAt(0)) 0L else avgTerminalExpressBillTotalCount.get(0).asInstanceOf[Number].longValue()
  )

  println(rowInfo)

  //将计算好的结果数据写入到结果对象中
  rows.append(rowInfo)

  //释放资源
  expressBillDetailByDayDF.unpersist()
  customerTypeExpressBillTotalCountDF.unpersist()
  dotExpressBillTotalCountDF.unpersist()
  channelExpressBillTotalCountDF.unpersist()
  terminalExpressBillTotalCountDF.unpersist()
})

//创建DataFrame：schema+rdd（数据）
//定义指标结果的schema信息
val schema: StructType = StructType(Array(
  StructField("id", StringType, false, Metadata.empty),
  StructField("totalExpressBillCount", LongType, false, Metadata.empty),
  StructField("maxTypeExpressBillTotalCount", LongType, false, Metadata.empty),
  StructField("minTypeExpressBillTotalCount", LongType, false, Metadata.empty),
  StructField("avgTypeExpressBillTotalCount", LongType, false, Metadata.empty),
  StructField("maxDotExpressBillTotalCount", LongType, false, Metadata.empty),
  StructField("minDotExpressBillTotalCount", LongType, false, Metadata.empty),
  StructField("avgDotExpressBillTotalCount", LongType, false, Metadata.empty),
  StructField("maxChannelExpressBillTotalCount", LongType, false, Metadata.empty),
  StructField("minChannelExpressBillTotalCount", LongType, false, Metadata.empty),
  StructField("avgChannelExpressBillTotalCount", LongType, false, Metadata.empty),
  StructField("maxTerminalExpressBillTotalCount", LongType, false, Metadata.empty),
  StructField("minTerminalExpressBillTotalCount", LongType, false, Metadata.empty),
  StructField("avgTerminalExpressBillTotalCount", LongType, false, Metadata.empty)
))

//将数据转换成rdd对象
val data: RDD[Row] = sparkSession.sparkContext.makeRDD(rows)

//根据表结构和数据创建DataFrame对象
val quotaDF: DataFrame = sparkSession.createDataFrame(data, schema)

//TODO 5）将计算好的指标数据写入到kudu数据库中
//将dataframe数据写入到kudu数据库
save(quotaDF, OfflineTableDefine.expressBillSummary)

//TODO 6）删除缓存数据
expressBillDetailGroupByDayDF.unpersist()
expressBillDetailDF.unpersist()

package cn.it.logistics.offline.dws

import cn.it.logistics.common.{Configuration, DateHelper, OfflineTableDefine, SparkUtils}
import cn.it.logistics.offline.OfflineApp
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SparkSession}
import org.apache.spark.sql.types.{DoubleType, LongType, Metadata, StringType, StructField, StructType}
import org.apache.spark.sql.functions._
import org.apache.spark.storage.StorageLevel

import scala.collection.mutable.ArrayBuffer

/**
 * 快递单主题指标开发
 */
object ExpressBillDWS extends OfflineApp{
  //定义应用程序的名称
  val appName = this.getClass.getSimpleName

  /**
   * 入口函数
   * @param args
   */
  def main(args: Array[String]): Unit = {
    /**
     * 实现步骤：
     * 1）创建SparkConf对象
     * 2）创建SparkSession对象
     * 3）读取快递明细宽表的数据
     * 4）对快递明细宽表的数据进行指标的计算
     * 5）将计算好的指标数据写入到kudu数据库中
     * 5.1：定义指标结果表的schema信息
     * 5.2：组织需要写入到kudu表的数据
     * 5.3：判断指标结果表是否存在，如果不存在则创建
     * 5.4：将数据写入到kudu表中
     * 6）删除缓存数据
     * 7）停止任务，退出sparksession
     */

    //TODO 1）创建SparkConf对象
    val sparkConf: SparkConf = SparkUtils.autoSettingEnv(
      SparkUtils.sparkConf(appName)
    )

    //TODO 2）创建SparkSession对象
    val sparkSession: SparkSession = SparkUtils.getSparkSession(sparkConf)
    sparkSession.sparkContext.setLogLevel(Configuration.LOG_OFF)

    //处理数据
    execute(sparkSession)
  }

  /**
   * 数据处理
   *
   * @param sparkSession
   */
  override def execute(sparkSession: SparkSession): Unit = {
    //TODO 3）读取快递单明细宽表的数据
    val expressBillDetailDF: DataFrame = getKuduSource(sparkSession, OfflineTableDefine.expressBillDetail, Configuration.isFirstRunnable)

    //读取出来的明细宽表数据可能是增量的数据，也可能是全量的数据
    //全量的数据是包含多个日期的数据，增量数据是前一天的数据
    //需要计算的指标是以日为单位，每天的最大快递单数、最小快递单数、平均快递单数据
    //因此需要对读取出来的快递单明细宽表数据按照日为单位进行分组，然后统计每日的指标数据
    val expressBillDetailGroupByDayDF: DataFrame = expressBillDetailDF.select("day").groupBy("day").count().cache()

    //导入隐式转换
    import sparkSession.implicits._

    //定义计算好的指标结果集合对象
    val rows: ArrayBuffer[Row] = ArrayBuffer[Row]()

    //循环遍历所有日期的数据
    expressBillDetailGroupByDayDF.collect().foreach(row=>{
      //获取到需要处理的数据所在的日期
      val day: String = row.getAs[String](0)

      //根据日期查询该日期内的快递单明细数据，然后将查询到的结果进行指标计算（指定日期的指标）
      val expressBillDetailByDayDF: DataFrame = expressBillDetailDF.where(col("day") === day).toDF().persist(StorageLevel.DISK_ONLY_2)

      //TODO 4）对快递明细宽表的数据进行指标的计算
      //总快递单数
      val totalExpressBillCount: Long = expressBillDetailByDayDF.agg(count("express_number")).first().getLong(0)
      //各类客户的快递单数
      val customerTypeExpressBillTotalCountDF: DataFrame = expressBillDetailByDayDF.groupBy($"type").agg(count("express_number").as("express_number")).cache()

      //各类客户最大快递单数
      val maxTypeExpressBillTotalCount: Row = customerTypeExpressBillTotalCountDF.agg(max("express_number")).first()

      //各类客户最小快递单数
      val minTypeExpressBillTotalCount: Row = customerTypeExpressBillTotalCountDF.agg(min("express_number")).first()

      //各类客户平均快递单数
      val avgTypeExpressBillTotalCount: Row = customerTypeExpressBillTotalCountDF.agg(avg("express_number")).first()

      //各网点的快递单数
      val dotExpressBillTotalCountDF: DataFrame = expressBillDetailByDayDF.groupBy($"dot_id").agg(count("express_number").as("express_number")).cache()

      //各网点最大快递单数
      val maxDotExpressBillTotalCount: Row = dotExpressBillTotalCountDF.agg(max("express_number")).first()

      //各网点最小快递单数
      val minDotExpressBillTotalCount: Row = dotExpressBillTotalCountDF.agg(min("express_number")).first()

      //各网点平均快递单数
      val avgDotExpressBillTotalCount: Row = dotExpressBillTotalCountDF.agg(avg("express_number")).first()

      //各渠道的快递单数
      val channelExpressBillTotalCountDF: DataFrame = expressBillDetailByDayDF.groupBy($"order_channel_id").agg(count("express_number").as("express_number")).cache()

      //各渠道最大快递单数
      val maxChannelExpressBillTotalCount: Row = channelExpressBillTotalCountDF.agg(max("express_number")).first()

      //各渠道最小快递单数
      val minChannelExpressBillTotalCount: Row = channelExpressBillTotalCountDF.agg(min("express_number")).first()

      //各渠道平均快递单数
      val avgChannelExpressBillTotalCount: Row = channelExpressBillTotalCountDF.agg(avg("express_number")).first()

      val terminalExpressBillTotalCountDF: DataFrame = expressBillDetailByDayDF.groupBy($"order_terminal_type").agg(count("express_number").as("express_number")).cache()
      println(terminalExpressBillTotalCountDF)

      //各终端最大快递单数
      val maxTerminalExpressBillTotalCount: Row = terminalExpressBillTotalCountDF.agg(max("express_number")).first()

      //各终端最小快递单数
      val minTerminalExpressBillTotalCount: Row = terminalExpressBillTotalCountDF.agg(min("express_number")).first()

      //各终端平均快递单数
      val avgTerminalExpressBillTotalCount: Row = terminalExpressBillTotalCountDF.agg(avg("express_number")).first()
      println(avgTerminalExpressBillTotalCount)

      //将每条记录写入到Row对象中
      val rowInfo = Row(
        day,
        totalExpressBillCount, //总快递单数
        if(maxTypeExpressBillTotalCount.isNullAt(0)) 0L else maxTypeExpressBillTotalCount.get(0).asInstanceOf[Number].longValue(),
        if(minTypeExpressBillTotalCount.isNullAt(0)) 0L else minTypeExpressBillTotalCount.get(0).asInstanceOf[Number].longValue(),
        if(avgTypeExpressBillTotalCount.isNullAt(0)) 0L else avgTypeExpressBillTotalCount.get(0).asInstanceOf[Number].longValue(),
        if(maxDotExpressBillTotalCount.isNullAt(0)) 0L else maxDotExpressBillTotalCount.get(0).asInstanceOf[Number].longValue(),
        if(minDotExpressBillTotalCount.isNullAt(0)) 0L else minDotExpressBillTotalCount.get(0).asInstanceOf[Number].longValue(),
        if(avgDotExpressBillTotalCount.isNullAt(0)) 0L else avgDotExpressBillTotalCount.get(0).asInstanceOf[Number].longValue(),
        if(maxChannelExpressBillTotalCount.isNullAt(0)) 0L else maxChannelExpressBillTotalCount.get(0).asInstanceOf[Number].longValue(),
        if(minChannelExpressBillTotalCount.isNullAt(0)) 0L else minChannelExpressBillTotalCount.get(0).asInstanceOf[Number].longValue(),
        if(avgChannelExpressBillTotalCount.isNullAt(0)) 0L else avgChannelExpressBillTotalCount.get(0).asInstanceOf[Number].longValue(),
        if(maxTerminalExpressBillTotalCount.isNullAt(0)) 0L else maxTerminalExpressBillTotalCount.get(0).asInstanceOf[Number].longValue(),
        if(minTerminalExpressBillTotalCount.isNullAt(0)) 0L else minTerminalExpressBillTotalCount.get(0).asInstanceOf[Number].longValue(),
        if(avgTerminalExpressBillTotalCount.isNullAt(0)) 0L else avgTerminalExpressBillTotalCount.get(0).asInstanceOf[Number].longValue()
      )

      println(rowInfo)

      //将计算好的结果数据写入到结果对象中
      rows.append(rowInfo)

      //释放资源
      expressBillDetailByDayDF.unpersist()
      customerTypeExpressBillTotalCountDF.unpersist()
      dotExpressBillTotalCountDF.unpersist()
      channelExpressBillTotalCountDF.unpersist()
      terminalExpressBillTotalCountDF.unpersist()
    })

    //创建DataFrame：schema+rdd（数据）
    //定义指标结果的schema信息
    val schema: StructType = StructType(Array(
      StructField("id", StringType, false, Metadata.empty),
      StructField("totalExpressBillCount", LongType, false, Metadata.empty),
      StructField("maxTypeExpressBillTotalCount", LongType, false, Metadata.empty),
      StructField("minTypeExpressBillTotalCount", LongType, false, Metadata.empty),
      StructField("avgTypeExpressBillTotalCount", LongType, false, Metadata.empty),
      StructField("maxDotExpressBillTotalCount", LongType, false, Metadata.empty),
      StructField("minDotExpressBillTotalCount", LongType, false, Metadata.empty),
      StructField("avgDotExpressBillTotalCount", LongType, false, Metadata.empty),
      StructField("maxChannelExpressBillTotalCount", LongType, false, Metadata.empty),
      StructField("minChannelExpressBillTotalCount", LongType, false, Metadata.empty),
      StructField("avgChannelExpressBillTotalCount", LongType, false, Metadata.empty),
      StructField("maxTerminalExpressBillTotalCount", LongType, false, Metadata.empty),
      StructField("minTerminalExpressBillTotalCount", LongType, false, Metadata.empty),
      StructField("avgTerminalExpressBillTotalCount", LongType, false, Metadata.empty)
    ))

    //将数据转换成rdd对象
    val data: RDD[Row] = sparkSession.sparkContext.makeRDD(rows)

    //根据表结构和数据创建DataFrame对象
    val quotaDF: DataFrame = sparkSession.createDataFrame(data, schema)

    //TODO 5）将计算好的指标数据写入到kudu数据库中
    //将dataframe数据写入到kudu数据库
    save(quotaDF, OfflineTableDefine.expressBillSummary)

    //TODO 6）删除缓存数据
    expressBillDetailGroupByDayDF.unpersist()
    expressBillDetailDF.unpersist()

    //TODO 7）停止任务，退出sparksession
    sparkSession.stop()
  }
}