客快物流大数据项目(六十五)：仓库主题

SELECT 
wsDF."id" AS id,
wsDF."pw_id" AS pw_id,
wsDF."pw_waybill_id" AS pw_waybill_id,
wsDF."pw_waybill_number" AS pw_waybill_number,
wsDF."ow_id" AS ow_id,
wsDF."ow_waybill_id" AS ow_waybill_id,
wsDF."ow_waybill_number" AS ow_waybill_number,
wsDF."sw_id" AS sw_id,
warehouseDF."name" AS sw_name,
wsDF."ew_id" AS ew_id,
wsDF."transport_tool_id" AS transport_tool_id,
wsDF."pw_driver1_id" AS pw_driver1_id,
wsDF."pw_driver2_id" AS pw_driver2_id,
wsDF."pw_driver3_id" AS pw_driver3_id,
wsDF."ow_driver1_id" AS ow_driver1_id,
wsDF."ow_driver2_id" AS ow_driver2_id,
wsDF."ow_driver3_id" AS ow_driver3_id,
wsDF."route_id" AS route_id,
wsDF."distance" AS distance,
wsDF."duration" AS duration,
wsDF."state" AS state,
wsDF."start_vehicle_dt" AS start_vehicle_dt,
wsDF."predict_arrivals_dt" AS predict_arrivals_dt,
wsDF."actual_arrivals_dt" AS actual_arrivals_dt,
wsDF."cdt" AS cdt,
wsDF."udt" AS udt,
wsDF."remark" AS remark,
companyDF."id" AS company_id,
companyDF."company_name" AS company_name,
areasDF."id" AS area_id,
areasDF."name" AS area_name,
wsDF ."pw_waybill_number",
expressBillDF."express_number",
expressBillDF."cid",
consumerSenderDF."ciid",
expressPackageDF."id" AS package_id,
expressPackageDF."name" AS package_name,
consumerDF."id" AS consumer_id,
consumerDF."name" AS consumer_name,
codesDF16."code" AS consumer_type,
codesDF16."code_desc" AS consumer_name_desc,
wsDF ."sw_id",
warehouseDF."id" ,
pwDF."warehouse_id",
pwDF."pw_dot_id",
dotDF."id" AS dot_id,
dotDF."dot_name" AS dot_name
FROM "tbl_transport_record" wsDF 																					--运输记录表
LEFT JOIN "tbl_company_warehouse_map" companyWarehouseMapDF ON wsDF."sw_id" =companyWarehouseMapDF."warehouse_id" 	--公司仓库关联表
LEFT JOIN "tbl_company" companyDF ON companyWarehouseMapDF."company_id" = companyDF ."id" 							--公司表
LEFT JOIN "tbl_areas" areasDF ON companyDF."city_id"= areasDF."id" 													--区域表
LEFT JOIN "tbl_waybill" waybillDF ON wsDF ."pw_waybill_number"=waybillDF."waybill_number"							--运单表
LEFT JOIN "tbl_express_bill" expressBillDF ON  expressBillDF."express_number"=waybillDF."express_bill_number" 		--快递单表
LEFT JOIN "tbl_consumer_sender_info" consumerSenderDF ON consumerSenderDF."ciid" = expressBillDF."cid" 				--客户寄件信息表
LEFT JOIN "tbl_express_package" expressPackageDF ON consumerSenderDF."pkg_id" = expressPackageDF."id" 				--快递包裹表
LEFT JOIN "tbl_customer" consumerDF ON consumerDF."id" =consumerSenderDF."ciid" 									--客户表
LEFT JOIN "tbl_codes" codesDF16 ON codesDF16 ."type" =16 AND consumerDF."type" = codesDF16 ."code" 					--物流系统码表
LEFT JOIN "tbl_warehouse" warehouseDF ON wsDF ."sw_id"=warehouseDF."id" 											--仓库表
LEFT JOIN "tbl_push_warehouse" pwDF ON pwDF."warehouse_id" =warehouseDF."id" 										--入库表
LEFT JOIN "tbl_dot" dotDF ON pwDF."pw_dot_id" =dotDF."id"

package cn.it.logistics.offline.dwd
import cn.it.logistics.common.{Configuration, SparkUtils}

import cn.it.logistics.offline.OfflineApp

import org.apache.spark.SparkConf

import org.apache.spark.sql.SparkSession
/**

仓库主题开发
将运输记录事实表的数据与维度表的数据进行拉宽操作，将拉宽后的数据写入到kudu数据库中

*/

object WareHouseDWD 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)：将缓存的数据删除掉
7）停止任务，释放sparksession对象

*/



//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()

}

}

//TODO 3）加载kudu中的事实表和维度表的数据（将加载后的数据进行缓存）

//加载运输工具表的数据

val recordDF: DataFrame = getKuduSource(sparkSession, TableMapping.transportRecord, Configuration.isFirstRunnable).toDF().persist(StorageLevel.DISK_ONLY_2)

//加载公司仓库关联表的数据

val companyWareHouseMapDF: DataFrame = getKuduSource(sparkSession, TableMapping.companyWarehouseMap, true).toDF().persist(StorageLevel.DISK_ONLY_2)
//加载公司表的数据

val companyDF: DataFrame = getKuduSource(sparkSession, TableMapping.company, true).toDF().persist(StorageLevel.DISK_ONLY_2)
//加载区域表的数据

val areasDF: DataFrame = getKuduSource(sparkSession, TableMapping.areas, true).toDF().persist(StorageLevel.DISK_ONLY_2)
//加载运单表的数据

val wayBillDF: DataFrame = getKuduSource(sparkSession, TableMapping.waybill, true).toDF().persist(StorageLevel.DISK_ONLY_2)
//加载快递单表的数据

val expressBillDF: DataFrame = getKuduSource(sparkSession, TableMapping.expressBill, true).toDF().persist(StorageLevel.DISK_ONLY_2)
//加载客户寄件信息表数据

val senderInfoDF: DataFrame = getKuduSource(sparkSession, TableMapping.consumerSenderInfo, true).toDF().persist(StorageLevel.DISK_ONLY_2)
//加载包裹表数据

val expressPackageDF: DataFrame = getKuduSource(sparkSession, TableMapping.expressPackage, true).toDF().persist(StorageLevel.DISK_ONLY_2)
//加载客户表数据

val customerDF: DataFrame = getKuduSource(sparkSession, TableMapping.customer, true).toDF().persist(StorageLevel.DISK_ONLY_2)
//加载物流码表数据

val codesDF: DataFrame = getKuduSource(sparkSession, TableMapping.codes, true).toDF().persist(StorageLevel.DISK_ONLY_2)
//加载仓库表数据

val warehouseDF: DataFrame = getKuduSource(sparkSession, TableMapping.warehouse, true).toDF().persist(StorageLevel.DISK_ONLY_2)
//加载入库数据

val phWarehouseDF: DataFrame = getKuduSource(sparkSession, TableMapping.pushWarehouse, true).toDF().persist(StorageLevel.DISK_ONLY_2)
//加载入库数据

val dotDF: DataFrame = getKuduSource(sparkSession, TableMapping.dot, true).toDF().persist(StorageLevel.DISK_ONLY_2)
//导入隐士转换

import sparkSession.implicits._
//客户类型表

val customerTypeDF: DataFrame = codesDF.where($"type" === CodeTypeMapping.CustomType).select(

"code".as("customerTypeCode"), "codeDesc".as("customerTypeName"))

//TODO 4）定义维度表与事实表的关联

val joinType = "left_outer"

val wsDetailDF: DataFrame = recordDF.join(companyWareHouseMapDF, recordDF.col("swId") === companyWareHouseMapDF.col("warehouseId"), joinType) // 转运记录表与公司仓库关联表关联

.join(companyDF, companyWareHouseMapDF.col("companyId") === companyDF.col("id"), joinType) //公司仓库关联表与公司表关联

.join(areasDF, companyDF.col("cityId") === areasDF.col("id"), joinType) //公司表与区域表关联

.join(wayBillDF, recordDF.col("pwWaybillNumber") === wayBillDF.col("waybillNumber"), joinType) //运单表与转运记录表关联

.join(expressBillDF, wayBillDF.col("expressBillNumber") === expressBillDF.col("expressNumber"), joinType) //运单表与快递单表关联

.join(senderInfoDF, expressBillDF.col("cid") === senderInfoDF.col("ciid"), joinType) //客户寄件信息表与快递单表关联

.join(expressPackageDF, senderInfoDF.col("pkgId") === expressPackageDF.col("id"), joinType) //客户寄件信息表与包裹表关联

.join(customerDF, senderInfoDF.col("ciid") === customerDF.col("id"), joinType) //客户寄件信息表与客户表关联

.join(customerTypeDF, customerDF.col("type") === customerTypeDF.col("customerTypeCode"), joinType) //客户表与客户类别表关联

.join(warehouseDF, recordDF.col("swId")===warehouseDF.col("id"), joinType) //转运记录表与仓库表关联

.join(phWarehouseDF, phWarehouseDF.col("warehouseId")=== warehouseDF.col("id"), joinType) //入库表与仓库表关联

.join(dotDF, dotDF("id")=== phWarehouseDF.col("pwDotId"), joinType)//转运记录表与网点表关联

.withColumn("day", date_format(recordDF("cdt"), "yyyyMMdd"))//虚拟列,可以根据这个日期列作为分区字段，可以保证同一天的数据保存在同一个分区中

.sort(recordDF.col("cdt").asc)

.select(

recordDF("id"),         //转运记录id

recordDF("pwId").as("pw_id"),   //入库表的id

recordDF("pwWaybillId").as("pw_waybill_id"),    //入库运单id

recordDF("pwWaybillNumber").as("pw_waybill_number"),  //入库运单编号

recordDF("owId").as("ow_id"),       //出库id

recordDF("owWaybillId").as("ow_waybill_id"),  //出库运单id

recordDF("owWaybillNumber").as("ow_waybill_number"),  //出库运单编号

recordDF("swId").as("sw_id"),     //起点仓库id

warehouseDF.col("name").as("sw_name"),  //起点仓库名称

recordDF("ewId").as("ew_id"),    //到达仓库id

recordDF("transportToolId").as("transport_tool_id"),  //运输工具id

recordDF("pwDriver1Id").as("pw_driver1_id"),    //入库车辆驾驶员

recordDF("pwDriver2Id").as("pw_driver2_id"),    //入库车辆驾驶员2

recordDF("pwDriver3Id").as("pw_driver3_id"),    //入库车辆驾驶员3

recordDF("owDriver1Id").as("ow_driver1_id"),    //出库车辆驾驶员

recordDF("owDriver2Id").as("ow_driver2_id"),    //出库车辆驾驶员2

recordDF("owDriver3Id").as("ow_driver3_id"),    //出库车辆驾驶员3

recordDF("routeId").as("route_id"),       //线路id

recordDF("distance").cast(IntegerType),   //运输里程

recordDF("duration").cast(IntegerType),    //运输耗时

recordDF("state").cast(IntegerType),      //转运状态id

recordDF("startVehicleDt").as("start_vehicle_dt"),    //发车时间

recordDF("predictArrivalsDt").as("predict_arrivals_dt"), //预计到达时间

recordDF("actualArrivalsDt").as("actual_arrivals_dt"),   //实际到达时间

recordDF("cdt"), //创建时间

recordDF("udt"), //修改时间

recordDF("remark"), //备注

companyDF("id").alias("company_id"), //公司id

companyDF("companyName").as("company_name"), //公司名称

areasDF("id").alias("area_id"),   //区域id

areasDF("name").alias("area_name"), //区域名称

expressPackageDF.col("id").alias("package_id"), //包裹id

expressPackageDF.col("name").alias("package_name"), //包裹名称

customerDF.col("id").alias("cid"),

customerDF.col("name").alias("cname"),

customerTypeDF("customerTypeCode").alias("ctype"),

customerTypeDF("customerTypeName").alias("ctype_name"),

dotDF("id").as("dot_id"),

dotDF("dotName").as("dot_name"),

$"day"

)

//TODO 5）将拉宽后的数据再次写回到kudu数据库中（DWD明细层）

save(wsDetailDF, OfflineTableDefine.wareHouseDetail)

//删除缓存，释放资源

companyWareHouseMapDF.unpersist()

companyDF.unpersist()

areasDF.unpersist()

wayBillDF.unpersist()

expressBillDF.unpersist()

senderInfoDF.unpersist()

expressPackageDF.unpersist()

customerDF.unpersist()

customerTypeDF.unpersist()

warehouseDF.unpersist()

phWarehouseDF.unpersist()

dotDF.unpersist()

recordDF.unpersist()

package cn.it.logistics.offline.dwd

import cn.it.logistics.common.{CodeTypeMapping, Configuration, OfflineTableDefine, SparkUtils, TableMapping}

import cn.it.logistics.offline.OfflineApp

import org.apache.spark.SparkConf

import org.apache.spark.sql.functions.date_format

import org.apache.spark.sql.types.IntegerType

import org.apache.spark.sql.{DataFrame, SparkSession}

import org.apache.spark.storage.StorageLevel
/**

仓库主题开发
将运输记录事实表的数据与维度表的数据进行拉宽操作，将拉宽后的数据写入到kudu数据库中

*/

object WareHouseDWD 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)：将缓存的数据删除掉
7）停止任务，释放sparksession对象

*/



//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中的事实表和维度表的数据（将加载后的数据进行缓存）

//加载运输工具表的数据

val recordDF: DataFrame = getKuduSource(sparkSession, TableMapping.transportRecord, Configuration.isFirstRunnable).toDF().persist(StorageLevel.DISK_ONLY_2)

//加载公司仓库关联表的数据
val companyWareHouseMapDF: DataFrame = getKuduSource(sparkSession, TableMapping.companyWarehouseMap, true).toDF().persist(StorageLevel.DISK_ONLY_2)

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

//加载区域表的数据
val areasDF: DataFrame = getKuduSource(sparkSession, TableMapping.areas, true).toDF().persist(StorageLevel.DISK_ONLY_2)

//加载运单表的数据
val wayBillDF: DataFrame = getKuduSource(sparkSession, TableMapping.waybill, true).toDF().persist(StorageLevel.DISK_ONLY_2)

//加载快递单表的数据
val expressBillDF: DataFrame = getKuduSource(sparkSession, TableMapping.expressBill, true).toDF().persist(StorageLevel.DISK_ONLY_2)

//加载客户寄件信息表数据
val senderInfoDF: DataFrame = getKuduSource(sparkSession, TableMapping.consumerSenderInfo, true).toDF().persist(StorageLevel.DISK_ONLY_2)

//加载包裹表数据
val expressPackageDF: DataFrame = getKuduSource(sparkSession, TableMapping.expressPackage, true).toDF().persist(StorageLevel.DISK_ONLY_2)

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

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

//加载仓库表数据
val warehouseDF: DataFrame = getKuduSource(sparkSession, TableMapping.warehouse, true).toDF().persist(StorageLevel.DISK_ONLY_2)

//加载入库数据
val phWarehouseDF: DataFrame = getKuduSource(sparkSession, TableMapping.pushWarehouse, true).toDF().persist(StorageLevel.DISK_ONLY_2)

//加载入库数据
val dotDF: DataFrame = getKuduSource(sparkSession, TableMapping.dot, true).toDF().persist(StorageLevel.DISK_ONLY_2)

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

//客户类型表
val customerTypeDF: DataFrame = codesDF.where($"type" === CodeTypeMapping.CustomType).select(
  $"code".as("customerTypeCode"), $"codeDesc".as("customerTypeName"))

//TODO 4）定义维度表与事实表的关联
val joinType = "left_outer"
val wsDetailDF: DataFrame = recordDF.join(companyWareHouseMapDF, recordDF.col("swId") === companyWareHouseMapDF.col("warehouseId"), joinType) // 转运记录表与公司仓库关联表关联
  .join(companyDF, companyWareHouseMapDF.col("companyId") === companyDF.col("id"), joinType) //公司仓库关联表与公司表关联
  .join(areasDF, companyDF.col("cityId") === areasDF.col("id"), joinType) //公司表与区域表关联
  .join(wayBillDF, recordDF.col("pwWaybillNumber") === wayBillDF.col("waybillNumber"), joinType) //运单表与转运记录表关联
  .join(expressBillDF, wayBillDF.col("expressBillNumber") === expressBillDF.col("expressNumber"), joinType) //运单表与快递单表关联
  .join(senderInfoDF, expressBillDF.col("cid") === senderInfoDF.col("ciid"), joinType) //客户寄件信息表与快递单表关联
  .join(expressPackageDF, senderInfoDF.col("pkgId") === expressPackageDF.col("id"), joinType) //客户寄件信息表与包裹表关联
  .join(customerDF, senderInfoDF.col("ciid") === customerDF.col("id"), joinType) //客户寄件信息表与客户表关联
  .join(customerTypeDF, customerDF.col("type") === customerTypeDF.col("customerTypeCode"), joinType) //客户表与客户类别表关联
  .join(warehouseDF, recordDF.col("swId")===warehouseDF.col("id"), joinType) //转运记录表与仓库表关联
  .join(phWarehouseDF, phWarehouseDF.col("warehouseId")=== warehouseDF.col("id"), joinType) //入库表与仓库表关联
  .join(dotDF, dotDF("id")=== phWarehouseDF.col("pwDotId"), joinType)//转运记录表与网点表关联
  .withColumn("day", date_format(recordDF("cdt"), "yyyyMMdd"))//虚拟列,可以根据这个日期列作为分区字段，可以保证同一天的数据保存在同一个分区中
  .sort(recordDF.col("cdt").asc)
  .select(
    recordDF("id"),         //转运记录id
    recordDF("pwId").as("pw_id"),   //入库表的id
    recordDF("pwWaybillId").as("pw_waybill_id"),    //入库运单id
    recordDF("pwWaybillNumber").as("pw_waybill_number"),  //入库运单编号
    recordDF("owId").as("ow_id"),       //出库id
    recordDF("owWaybillId").as("ow_waybill_id"),  //出库运单id
    recordDF("owWaybillNumber").as("ow_waybill_number"),  //出库运单编号
    recordDF("swId").as("sw_id"),     //起点仓库id
    warehouseDF.col("name").as("sw_name"),  //起点仓库名称
    recordDF("ewId").as("ew_id"),    //到达仓库id
    recordDF("transportToolId").as("transport_tool_id"),  //运输工具id
    recordDF("pwDriver1Id").as("pw_driver1_id"),    //入库车辆驾驶员
    recordDF("pwDriver2Id").as("pw_driver2_id"),    //入库车辆驾驶员2
    recordDF("pwDriver3Id").as("pw_driver3_id"),    //入库车辆驾驶员3
    recordDF("owDriver1Id").as("ow_driver1_id"),    //出库车辆驾驶员
    recordDF("owDriver2Id").as("ow_driver2_id"),    //出库车辆驾驶员2
    recordDF("owDriver3Id").as("ow_driver3_id"),    //出库车辆驾驶员3
    recordDF("routeId").as("route_id"),       //线路id
    recordDF("distance").cast(IntegerType),   //运输里程
    recordDF("duration").cast(IntegerType),    //运输耗时
    recordDF("state").cast(IntegerType),      //转运状态id
    recordDF("startVehicleDt").as("start_vehicle_dt"),    //发车时间
    recordDF("predictArrivalsDt").as("predict_arrivals_dt"), //预计到达时间
    recordDF("actualArrivalsDt").as("actual_arrivals_dt"),   //实际到达时间
    recordDF("cdt"), //创建时间
    recordDF("udt"), //修改时间
    recordDF("remark"), //备注
    companyDF("id").alias("company_id"), //公司id
    companyDF("companyName").as("company_name"), //公司名称
    areasDF("id").alias("area_id"),   //区域id
    areasDF("name").alias("area_name"), //区域名称
    expressPackageDF.col("id").alias("package_id"), //包裹id
    expressPackageDF.col("name").alias("package_name"), //包裹名称
    customerDF.col("id").alias("cid"),
    customerDF.col("name").alias("cname"),
    customerTypeDF("customerTypeCode").alias("ctype"),
    customerTypeDF("customerTypeName").alias("ctype_name"),
    dotDF("id").as("dot_id"),
    dotDF("dotName").as("dot_name"),
    $"day"
  )

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

//删除缓存，释放资源
companyWareHouseMapDF.unpersist()
companyDF.unpersist()
areasDF.unpersist()
wayBillDF.unpersist()
expressBillDF.unpersist()
senderInfoDF.unpersist()
expressPackageDF.unpersist()
customerDF.unpersist()
customerTypeDF.unpersist()
warehouseDF.unpersist()
phWarehouseDF.unpersist()
dotDF.unpersist()
recordDF.unpersist()

sparkSession.stop()

}

}

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
/**

仓库主题指标开发
将计算好的指标数据结果写入到kudu数据库中

*/

object WarehouseDWS  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 = {

sparkSession.stop()

}

}

//TODO 3）读取仓库明细宽表的数据

val whDetailDF: DataFrame = getKuduSource(sparkSession, OfflineTableDefine.wareHouseDetail, Configuration.isFirstRunnable).toDF().persist(StorageLevel.DISK_ONLY_2)

//根据仓库的日期进行分组

val whDetailGroupByDayDF: DataFrame = whDetailDF.select("day").groupBy("day").count().cache()

//导入隐式转换

import sparkSession.implicits._
//定义计算好的指标结果集合对象

val rows: ArrayBuffer[Row] = ArrayBufferRow
//循环遍历每个日期的仓库明细宽表数据

whDetailGroupByDayDF.collect().foreach(row=>{

//获取到要处理的数据所在的日期

val day: String = row.getAsString
//返回指定日期的仓库明细数据

val whDetailByDayDF: DataFrame = whDetailDF.where(col("day") === day).toDF().persist(StorageLevel.DISK_ONLY_2)
//TODO 4）对仓库明细宽表的数据进行指标的计算

//各仓库发车次数（从哪个仓库发出的货物，根据发出仓库的仓库id进行分组）

val wsTotalCountDF: DataFrame = whDetailByDayDF.groupBy("sw_id").agg(count("id").alias("cnt")).cache()
//各仓库最大发车次数

val maxAndMinAndAvgTotalCountDF: Row = wsTotalCountDF.agg(max("cnt"), min("cnt"), avg($"cnt")).first()

val wsMaxTotalCount: Any = maxAndMinAndAvgTotalCountDF(0)

//各仓库最小发车次数

val wsMinTotalCount: Any = maxAndMinAndAvgTotalCountDF(1)

//各仓库平均发车次数

val wsAvgTotalCount: Any = maxAndMinAndAvgTotalCountDF(2)
//各网点发车次数

val dotTotalCountDF: DataFrame = whDetailByDayDF.groupBy("dot_id").agg(count("id").alias("cnt")).cache()
//各网点最大发车次数

val maxAndMinAndAvgDotTotalCountDF: Row = dotTotalCountDF.agg(max("cnt"), min("cnt"), avg($"cnt")).first()

val dotMaxTotalCount: Any = maxAndMinAndAvgDotTotalCountDF(0)

//各网点最小发车次数

val dotMinTotalCount: Any = maxAndMinAndAvgDotTotalCountDF(1)

//各网点平均发车次数

val dotAvgTotalCount: Any = maxAndMinAndAvgDotTotalCountDF(2)
//各线路发车次数

val routeTotalCountDF: DataFrame = whDetailByDayDF.groupBy("route_id").agg(count("id").alias("cnt")).cache()

//各线路最大发车次数

val maxAndMinAndAvgRouteTotalCountDF: Row = routeTotalCountDF.agg(max("cnt"), min("cnt"), avg($"cnt")).first()

val routeMaxTotalCount: Any = maxAndMinAndAvgRouteTotalCountDF(0)

//各线路最小发车次数

val routeMinTotalCount: Any = maxAndMinAndAvgRouteTotalCountDF(1)

//各线路平均发车次数

val routeAvgTotalCount: Any = maxAndMinAndAvgRouteTotalCountDF(2)
//各类型客户发车次数

val cTypeCountDF: DataFrame = whDetailByDayDF.groupBy("ctype").agg(count("id").alias("cnt")).cache()
//各类型客户最大发车次数

val maxAndMinAndAvgCtypeTotalCountDF: Row = cTypeCountDF.agg(max("cnt"), min("cnt"), avg($"cnt")).first()

val consumerTypeMaxTotalCount: Any = maxAndMinAndAvgCtypeTotalCountDF(0)

//各类型客户最小发车次数

val consumerTypeMinTotalCount: Any = maxAndMinAndAvgCtypeTotalCountDF(1)

//各类型客户平均发车次数

val consumerTypeAvgTotalCount: Any = maxAndMinAndAvgCtypeTotalCountDF(2)
//各类型客户发车次数

val pkgCountDF: DataFrame = whDetailByDayDF.groupBy("package_id").agg(count("id").alias("cnt")).cache()
//各类型包裹最大发车次数

val maxAndMinAndAvgPkgTotalCountDF: Row = pkgCountDF.agg(max("cnt"), min("cnt"), avg($"cnt")).first()

val packageMaxTotalCount: Any = maxAndMinAndAvgPkgTotalCountDF(0)

//各类型包裹最小发车次数

val packageMinTotalCount: Any = maxAndMinAndAvgPkgTotalCountDF(1)

//各类型包裹平均发车次数

val packageAvgTotalCount: Any = maxAndMinAndAvgPkgTotalCountDF(2)
//各区域发车次数

val areaCountDF: DataFrame = whDetailByDayDF.groupBy("area_id").agg(count("id").alias("cnt")).cache()
//各区域最大发车次数

val maxAndMinAndAvgAreaTotalCountDF: Row = areaCountDF.agg(max("cnt"), min("cnt"), avg($"cnt")).first()

val areaMaxTotalCount: Any = maxAndMinAndAvgAreaTotalCountDF(0)

//各区域最小发车次数

val areaMinTotalCount: Any = maxAndMinAndAvgAreaTotalCountDF(1)

//各区域平均发车次数

val areaAvgTotalCount: Any = maxAndMinAndAvgAreaTotalCountDF(2)
//各公司发车次数

val companyCountDF: DataFrame = whDetailByDayDF.groupBy("company_id").agg(count("id").alias("cnt")).cache()
//各公司最大发车次数

val maxAndMinAndAvgCompanyTotalCountDF: Row = areaCountDF.agg(max("cnt"), min("cnt"), avg($"cnt")).first()

val companyMaxTotalCount: Any = maxAndMinAndAvgCompanyTotalCountDF(0)

//各公司最小发车次数

val companyMinTotalCount: Any = maxAndMinAndAvgCompanyTotalCountDF(1)

//各公司平均发车次数

val companyAvgTotalCount: Any = maxAndMinAndAvgCompanyTotalCountDF(2)
//将计算好的指标数据封装到row对象中

val rowInfo: Row = Row(

day,

if(wsMaxTotalCount==null) 0L else wsMaxTotalCount.asInstanceOf[Number].longValue(),

if(wsMinTotalCount==null) 0L else wsMinTotalCount.asInstanceOf[Number].longValue(),

if(wsAvgTotalCount==null) 0L else wsAvgTotalCount.asInstanceOf[Number].longValue(),

if(dotMaxTotalCount==null) 0L else dotMaxTotalCount.asInstanceOf[Number].longValue(),

if(dotMinTotalCount==null) 0L else dotMinTotalCount.asInstanceOf[Number].longValue(),

if(dotAvgTotalCount==null) 0L else dotAvgTotalCount.asInstanceOf[Number].longValue(),

if(routeMaxTotalCount==null) 0L else routeMaxTotalCount.asInstanceOf[Number].longValue(),

if(routeMinTotalCount==null) 0L else routeMinTotalCount.asInstanceOf[Number].longValue(),

if(routeAvgTotalCount==null) 0L else routeAvgTotalCount.asInstanceOf[Number].longValue(),

if(consumerTypeMaxTotalCount==null) 0L else consumerTypeMaxTotalCount.asInstanceOf[Number].longValue(),

if(consumerTypeMinTotalCount==null) 0L else consumerTypeMinTotalCount.asInstanceOf[Number].longValue(),

if(consumerTypeAvgTotalCount==null) 0L else consumerTypeAvgTotalCount.asInstanceOf[Number].longValue(),

if(packageMaxTotalCount==null) 0L else packageMaxTotalCount.asInstanceOf[Number].longValue(),

if(packageMinTotalCount==null) 0L else packageMinTotalCount.asInstanceOf[Number].longValue(),

if(packageAvgTotalCount==null) 0L else packageAvgTotalCount.asInstanceOf[Number].longValue(),

if(areaMaxTotalCount==null) 0L else areaMaxTotalCount.asInstanceOf[Number].longValue(),

if(areaMinTotalCount==null) 0L else areaMinTotalCount.asInstanceOf[Number].longValue(),

if(areaAvgTotalCount==null) 0L else areaAvgTotalCount.asInstanceOf[Number].longValue(),

if(companyMaxTotalCount==null) 0L else companyMaxTotalCount.asInstanceOf[Number].longValue(),

if(companyMinTotalCount==null) 0L else companyMinTotalCount.asInstanceOf[Number].longValue(),

if(companyAvgTotalCount==null) 0L else companyAvgTotalCount.asInstanceOf[Number].longValue()

)
rows.append(rowInfo)

println(rowInfo)
//删除缓存，释放资源

whDetailByDayDF.unpersist()

wsTotalCountDF.unpersist()

companyCountDF.unpersist()

areaCountDF.unpersist()

pkgCountDF.unpersist()

cTypeCountDF.unpersist()

routeTotalCountDF.unpersist()

dotTotalCountDF.unpersist()

})

//定义表结构信息

val schema = StructType(Array(

StructField("id", StringType, false, Metadata.empty),

StructField("wsMaxTotalCount", LongType, true, Metadata.empty),           //各仓库最大发车次数

StructField("wsMinTotalCount", LongType, true, Metadata.empty),           //各仓库最小发车次数

StructField("wsAvgTotalCount", LongType, true, Metadata.empty),           //各仓库平均发车次数

StructField("dotMaxTotalCount", LongType, true, Metadata.empty),          //各网点最大发车次数

StructField("dotMinTotalCount", LongType, true, Metadata.empty),          //各网点最小发车次数

StructField("dotAvgTotalCount", LongType, true, Metadata.empty),          //各网点平均发车次数

StructField("routeMaxTotalCount", LongType, true, Metadata.empty),        //各线路最大发车次数

StructField("routeMinTotalCount", LongType, true, Metadata.empty),        //各线路最小发车次数

StructField("routeAvgTotalCount", LongType, true, Metadata.empty),        //各线路平均发车次数

StructField("consumerTypeMaxTotalCount", LongType, true, Metadata.empty), //各客户类型最大发车次数

StructField("consumerTypeMinTotalCount", LongType, true, Metadata.empty), //各客户类型最小发车次数

StructField("consumerTypeAvgTotalCount", LongType, true, Metadata.empty), //各客户类型平均发车次数

StructField("packageMaxTotalCount", LongType, true, Metadata.empty),      //各类型包裹最大发车次数

StructField("packageMinTotalCount", LongType, true, Metadata.empty),      //各类型包裹最小发车次数

StructField("packageAvgTotalCount", LongType, true, Metadata.empty),      //各类型包裹平均发车次数

StructField("areaMaxTotalCount", LongType, true, Metadata.empty),         //各区域最大发车次数

StructField("areaMinTotalCount", LongType, true, Metadata.empty),         //各区域最小发车次数

StructField("areaAvgTotalCount", LongType, true, Metadata.empty),         //各区域平均发车次数

StructField("companyMaxTotalCount", LongType, true, Metadata.empty),      //各公司最大发车次数

StructField("companyMinTotalCount", LongType, true, Metadata.empty),      //各公司最小发车次数

StructField("companyAvgTotalCount", LongType, true, Metadata.empty)       //各公司平均发车次数

))

//将rows对象转换成rdd对象

val data: RDD[Row] = sparkSession.sparkContext.makeRDD(rows)

//使用rdd和schema创建dataFrame对象

val quotaDF: DataFrame = sparkSession.createDataFrame(data, schema)
//quotaDF.show()

//将数据写入到kudu数据库

save(quotaDF, OfflineTableDefine.wareHouseSummary)

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
/**

仓库主题指标开发
将计算好的指标数据结果写入到kudu数据库中

*/

object WarehouseDWS  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 whDetailDF: DataFrame = getKuduSource(sparkSession, OfflineTableDefine.wareHouseDetail, Configuration.isFirstRunnable).toDF().persist(StorageLevel.DISK_ONLY_2)

//根据仓库的日期进行分组
val whDetailGroupByDayDF: DataFrame = whDetailDF.select("day").groupBy("day").count().cache()

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

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

//循环遍历每个日期的仓库明细宽表数据
whDetailGroupByDayDF.collect().foreach(row=>{
  //获取到要处理的数据所在的日期
  val day: String = row.getAs[String](0)

  //返回指定日期的仓库明细数据
  val whDetailByDayDF: DataFrame = whDetailDF.where(col("day") === day).toDF().persist(StorageLevel.DISK_ONLY_2)

  //TODO 4）对仓库明细宽表的数据进行指标的计算
  //各仓库发车次数（从哪个仓库发出的货物，根据发出仓库的仓库id进行分组）
  val wsTotalCountDF: DataFrame = whDetailByDayDF.groupBy("sw_id").agg(count("id").alias("cnt")).cache()

  //各仓库最大发车次数
  val maxAndMinAndAvgTotalCountDF: Row = wsTotalCountDF.agg(max($"cnt"), min($"cnt"), avg($"cnt")).first()
  val wsMaxTotalCount: Any = maxAndMinAndAvgTotalCountDF(0)
  //各仓库最小发车次数
  val wsMinTotalCount: Any = maxAndMinAndAvgTotalCountDF(1)
  //各仓库平均发车次数
  val wsAvgTotalCount: Any = maxAndMinAndAvgTotalCountDF(2)

  //各网点发车次数
  val dotTotalCountDF: DataFrame = whDetailByDayDF.groupBy("dot_id").agg(count("id").alias("cnt")).cache()

  //各网点最大发车次数
  val maxAndMinAndAvgDotTotalCountDF: Row = dotTotalCountDF.agg(max($"cnt"), min($"cnt"), avg($"cnt")).first()
  val dotMaxTotalCount: Any = maxAndMinAndAvgDotTotalCountDF(0)
  //各网点最小发车次数
  val dotMinTotalCount: Any = maxAndMinAndAvgDotTotalCountDF(1)
  //各网点平均发车次数
  val dotAvgTotalCount: Any = maxAndMinAndAvgDotTotalCountDF(2)

  //各线路发车次数
  val routeTotalCountDF: DataFrame = whDetailByDayDF.groupBy("route_id").agg(count("id").alias("cnt")).cache()
  //各线路最大发车次数
  val maxAndMinAndAvgRouteTotalCountDF: Row = routeTotalCountDF.agg(max($"cnt"), min($"cnt"), avg($"cnt")).first()
  val routeMaxTotalCount: Any = maxAndMinAndAvgRouteTotalCountDF(0)
  //各线路最小发车次数
  val routeMinTotalCount: Any = maxAndMinAndAvgRouteTotalCountDF(1)
  //各线路平均发车次数
  val routeAvgTotalCount: Any = maxAndMinAndAvgRouteTotalCountDF(2)

  //各类型客户发车次数
  val cTypeCountDF: DataFrame = whDetailByDayDF.groupBy("ctype").agg(count("id").alias("cnt")).cache()

  //各类型客户最大发车次数
  val maxAndMinAndAvgCtypeTotalCountDF: Row = cTypeCountDF.agg(max($"cnt"), min($"cnt"), avg($"cnt")).first()
  val consumerTypeMaxTotalCount: Any = maxAndMinAndAvgCtypeTotalCountDF(0)
  //各类型客户最小发车次数
  val consumerTypeMinTotalCount: Any = maxAndMinAndAvgCtypeTotalCountDF(1)
  //各类型客户平均发车次数
  val consumerTypeAvgTotalCount: Any = maxAndMinAndAvgCtypeTotalCountDF(2)

  //各类型客户发车次数
  val pkgCountDF: DataFrame = whDetailByDayDF.groupBy("package_id").agg(count("id").alias("cnt")).cache()

  //各类型包裹最大发车次数
  val maxAndMinAndAvgPkgTotalCountDF: Row = pkgCountDF.agg(max($"cnt"), min($"cnt"), avg($"cnt")).first()
  val packageMaxTotalCount: Any = maxAndMinAndAvgPkgTotalCountDF(0)
  //各类型包裹最小发车次数
  val packageMinTotalCount: Any = maxAndMinAndAvgPkgTotalCountDF(1)
  //各类型包裹平均发车次数
  val packageAvgTotalCount: Any = maxAndMinAndAvgPkgTotalCountDF(2)

  //各区域发车次数
  val areaCountDF: DataFrame = whDetailByDayDF.groupBy("area_id").agg(count("id").alias("cnt")).cache()

  //各区域最大发车次数
  val maxAndMinAndAvgAreaTotalCountDF: Row = areaCountDF.agg(max($"cnt"), min($"cnt"), avg($"cnt")).first()
  val areaMaxTotalCount: Any = maxAndMinAndAvgAreaTotalCountDF(0)
  //各区域最小发车次数
  val areaMinTotalCount: Any = maxAndMinAndAvgAreaTotalCountDF(1)
  //各区域平均发车次数
  val areaAvgTotalCount: Any = maxAndMinAndAvgAreaTotalCountDF(2)

  //各公司发车次数
  val companyCountDF: DataFrame = whDetailByDayDF.groupBy("company_id").agg(count("id").alias("cnt")).cache()

  //各公司最大发车次数
  val maxAndMinAndAvgCompanyTotalCountDF: Row = areaCountDF.agg(max($"cnt"), min($"cnt"), avg($"cnt")).first()
  val companyMaxTotalCount: Any = maxAndMinAndAvgCompanyTotalCountDF(0)
  //各公司最小发车次数
  val companyMinTotalCount: Any = maxAndMinAndAvgCompanyTotalCountDF(1)
  //各公司平均发车次数
  val companyAvgTotalCount: Any = maxAndMinAndAvgCompanyTotalCountDF(2)

  //将计算好的指标数据封装到row对象中
  val rowInfo: Row = Row(
    day,
    if(wsMaxTotalCount==null) 0L else wsMaxTotalCount.asInstanceOf[Number].longValue(),
    if(wsMinTotalCount==null) 0L else wsMinTotalCount.asInstanceOf[Number].longValue(),
    if(wsAvgTotalCount==null) 0L else wsAvgTotalCount.asInstanceOf[Number].longValue(),
    if(dotMaxTotalCount==null) 0L else dotMaxTotalCount.asInstanceOf[Number].longValue(),
    if(dotMinTotalCount==null) 0L else dotMinTotalCount.asInstanceOf[Number].longValue(),
    if(dotAvgTotalCount==null) 0L else dotAvgTotalCount.asInstanceOf[Number].longValue(),
    if(routeMaxTotalCount==null) 0L else routeMaxTotalCount.asInstanceOf[Number].longValue(),
    if(routeMinTotalCount==null) 0L else routeMinTotalCount.asInstanceOf[Number].longValue(),
    if(routeAvgTotalCount==null) 0L else routeAvgTotalCount.asInstanceOf[Number].longValue(),
    if(consumerTypeMaxTotalCount==null) 0L else consumerTypeMaxTotalCount.asInstanceOf[Number].longValue(),
    if(consumerTypeMinTotalCount==null) 0L else consumerTypeMinTotalCount.asInstanceOf[Number].longValue(),
    if(consumerTypeAvgTotalCount==null) 0L else consumerTypeAvgTotalCount.asInstanceOf[Number].longValue(),
    if(packageMaxTotalCount==null) 0L else packageMaxTotalCount.asInstanceOf[Number].longValue(),
    if(packageMinTotalCount==null) 0L else packageMinTotalCount.asInstanceOf[Number].longValue(),
    if(packageAvgTotalCount==null) 0L else packageAvgTotalCount.asInstanceOf[Number].longValue(),
    if(areaMaxTotalCount==null) 0L else areaMaxTotalCount.asInstanceOf[Number].longValue(),
    if(areaMinTotalCount==null) 0L else areaMinTotalCount.asInstanceOf[Number].longValue(),
    if(areaAvgTotalCount==null) 0L else areaAvgTotalCount.asInstanceOf[Number].longValue(),
    if(companyMaxTotalCount==null) 0L else companyMaxTotalCount.asInstanceOf[Number].longValue(),
    if(companyMinTotalCount==null) 0L else companyMinTotalCount.asInstanceOf[Number].longValue(),
    if(companyAvgTotalCount==null) 0L else companyAvgTotalCount.asInstanceOf[Number].longValue()
  )

  rows.append(rowInfo)
  println(rowInfo)

  //删除缓存，释放资源
  whDetailByDayDF.unpersist()
  wsTotalCountDF.unpersist()
  companyCountDF.unpersist()
  areaCountDF.unpersist()
  pkgCountDF.unpersist()
  cTypeCountDF.unpersist()
  routeTotalCountDF.unpersist()
  dotTotalCountDF.unpersist()
})

//定义表结构信息
val schema = StructType(Array(
  StructField("id", StringType, false, Metadata.empty),
  StructField("wsMaxTotalCount", LongType, true, Metadata.empty),           //各仓库最大发车次数
  StructField("wsMinTotalCount", LongType, true, Metadata.empty),           //各仓库最小发车次数
  StructField("wsAvgTotalCount", LongType, true, Metadata.empty),           //各仓库平均发车次数
  StructField("dotMaxTotalCount", LongType, true, Metadata.empty),          //各网点最大发车次数
  StructField("dotMinTotalCount", LongType, true, Metadata.empty),          //各网点最小发车次数
  StructField("dotAvgTotalCount", LongType, true, Metadata.empty),          //各网点平均发车次数
  StructField("routeMaxTotalCount", LongType, true, Metadata.empty),        //各线路最大发车次数
  StructField("routeMinTotalCount", LongType, true, Metadata.empty),        //各线路最小发车次数
  StructField("routeAvgTotalCount", LongType, true, Metadata.empty),        //各线路平均发车次数
  StructField("consumerTypeMaxTotalCount", LongType, true, Metadata.empty), //各客户类型最大发车次数
  StructField("consumerTypeMinTotalCount", LongType, true, Metadata.empty), //各客户类型最小发车次数
  StructField("consumerTypeAvgTotalCount", LongType, true, Metadata.empty), //各客户类型平均发车次数
  StructField("packageMaxTotalCount", LongType, true, Metadata.empty),      //各类型包裹最大发车次数
  StructField("packageMinTotalCount", LongType, true, Metadata.empty),      //各类型包裹最小发车次数
  StructField("packageAvgTotalCount", LongType, true, Metadata.empty),      //各类型包裹平均发车次数
  StructField("areaMaxTotalCount", LongType, true, Metadata.empty),         //各区域最大发车次数
  StructField("areaMinTotalCount", LongType, true, Metadata.empty),         //各区域最小发车次数
  StructField("areaAvgTotalCount", LongType, true, Metadata.empty),         //各区域平均发车次数
  StructField("companyMaxTotalCount", LongType, true, Metadata.empty),      //各公司最大发车次数
  StructField("companyMinTotalCount", LongType, true, Metadata.empty),      //各公司最小发车次数
  StructField("companyAvgTotalCount", LongType, true, Metadata.empty)       //各公司平均发车次数
))

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

//使用rdd和schema创建dataFrame对象
val quotaDF: DataFrame = sparkSession.createDataFrame(data, schema)

//quotaDF.show()
//将数据写入到kudu数据库
save(quotaDF, OfflineTableDefine.wareHouseSummary)

sparkSession.stop()

}

}