客快物流大数据项目(六十二)：主题及指标开发 common包下定义的一些内容 一般有用 看1

主题及指标开发

一、主题开发业务流程

二、离线模块初始化

1、创建包结构

2、​​​​​​​创建时间处理工具

3、​​​​​​​定义主题宽表及指标结果表的表名

4、​​​​​​​物流字典码表数据类型定义枚举类

5、​​​​​​​封装公共接口

主题及指标开发

一、主题开发业务流程

二、​​​​​​​离线模块初始化

1、​​​​​​​创建包结构

本次项目采用scala编程语言，因此创建scala目录

2、​​​​​​​创建时间处理工具

实现步骤：

• 在公共模块的scala目录下的common程序包下创建DateHelper对象
  • 实现获取当前日期
  • 实现获取昨天日期

package cn.it.logistics.common

import java.text.SimpleDateFormat
import java.util.Date

/**
 * 时间处理工具类
 */
object DateHelper {
  /**
   * 返回昨天的时间
   */
  def getyesterday(format:String)={
    //当前时间减去一天（昨天时间）
    new SimpleDateFormat(format).format(new Date(System.currentTimeMillis() - 1000 * 60 * 60 * 24))
  }

  /**
   * 返回今天的时间
   * @param format
   */
  def gettoday(format:String) = {
    //获取指定格式的当前时间
    new SimpleDateFormat(format).format(new Date)
  }
}

3、​​​​​​​定义主题宽表及指标结果表的表名

每个主题都需要拉宽操作将拉宽后的数据存储到kudu表中，同时指标计算的数据最终也需要落地到kudu表，因此提前将各个主题相关表名定义出来

实现步骤：

• 在公共模块的scala目录下的common程序包下创建OfflineTableDefine单例对象
• 定义各个主题相关的表名

参考代码：

package cn.it.logistics.common

/**
 * 自定义离线计算结果表
 */
object OfflineTableDefine {
  //快递单明细表
  val expressBillDetail  = "tbl_express_bill_detail"
  //快递单指标结果表
  val expressBillSummary = "tbl_express_bill_summary"
  //运单明细表
  val wayBillDetail = "tbl_waybill_detail"
  //运单指标结果表
  val wayBillSummary = "tbl_waybill_summary"
  //仓库明细表
  val wareHouseDetail = "tbl_warehouse_detail"
  //仓库指标结果表
  val wareHouseSummary = "tbl_warehouse_summary"
  //网点车辆明细表
  val dotTransportToolDetail = "tbl_dot_transport_tool_detail"
  //仓库车辆明细表
  val warehouseTransportToolDetail = "tbl_warehouse_transport_tool_detail"
  //网点车辆指标结果表
  val ttDotSummary = "tbl_dot_transport_tool_summary"
  //仓库车辆指标结果表
  val ttWsSummary = "tbl_warehouse_transport_tool_summary"
  //客户明细表数据
  val customerDetail = "tbl_customer_detail"
  //客户指标结果表数据
  val customerSummery = "tbl_customer_summary"
}

4、​​​​​​​物流字典码表数据类型定义枚举类

为了后续使用方便且易于维护，根据物流字典表的数据类型定义成枚举工具类，物流字典表的数据如下：

来自：tbl_codes表

实现步骤：

• 在公共模块的scala目录下的common程序包下创建CodeTypeMapping对象
• 根据物流字典表数据类型定义属性

实现过程：

• 在公共模块的scala目录下的common程序包下创建CodeTypeMapping对象
• 根据物流字典表数据类型定义属性

package cn.it.logistics.common

/**
 * 定义物流字典编码类型映射工具类
 */
class CodeTypeMapping {
  //注册渠道
  val RegisterChannel = 1
  //揽件状态
  val CollectStatus = 2
  //派件状态
  val DispatchStatus = 3
  //快递员状态
  val CourierStatus = 4
  //地址类型
  val AddressType = 5
  //网点状态
  val DotStatus = 6
  //员工状态
  val StaffStatus = 7
  //是否保价
  val IsInsured = 8
  //运输工具类型
  val TransportType = 9
  //运输工具状态
  val TransportStatus = 10
  //仓库类型
  val WareHouseType = 11
  //是否租赁
  val IsRent = 12
  //货架状态
  val GoodsShelvesStatue = 13
  //回执单状态
  val ReceiptStatus = 14
  //出入库类型
  val WarehousingType = 15
  //客户类型
  val CustomType = 16
  //下单终端类型
  val OrderTerminalType = 17
  //下单渠道类型
  val OrderChannelType = 18
}
object CodeTypeMapping extends CodeTypeMapping{
}

5、​​​​​​​封装公共接口

根据分析：主题开发数据的来源都是来自于kudu数据库，将数据进行拉宽或者将计算好的指标最终需要写入到kudu表中，因此根据以上流程抽象出来公共接口

实现步骤：

• 在offline目录下创建OfflineApp单例对象
  • 定义数据的读取方法：getKuduSource
  • 定义数据的处理方法：execute
  • 定义数据的存储方法：save

参考代码：

package cn.it.logistics.offline

import cn.it.logistics.common.{Configuration, DateHelper, Tools}
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
import org.apache.spark.sql.functions.{col, date_format}

/**
 * 根据不同的主题开发定义抽象方法
 * 1）数据读取
 * 2）数据处理
 * 3）数据保存
 */
trait OfflineApp {
  /**
   * 读取kudu表的数据
   * @param sparkSession
   * @param tableName
   * @param isLoadFullData
   */
  def getKuduSource(sparkSession: SparkSession, tableName:String, isLoadFullData:Boolean = false)= {
    if (isLoadFullData) {
      //加载全部的数据
      sparkSession.read.format(Configuration.SPARK_KUDU_FORMAT).options(
        Map(
          "kudu.master" -> Configuration.kuduRpcAddress,
          "kudu.table" -> tableName,
          "kudu.socketReadTimeoutMs"-> "60000")
      ).load().toDF()
    } else {
      //加载增量数据
      sparkSession.read.format(Configuration.SPARK_KUDU_FORMAT).options(
        Map(
          "kudu.master" -> Configuration.kuduRpcAddress,
          "kudu.table" -> tableName,
          "kudu.socketReadTimeoutMs"-> "60000")
      ).load()
        .where(date_format(col("cdt"), "yyyyMMdd") === DateHelper.getyesterday("yyyyMMdd")).toDF()
    }
  }

  /**
   * 数据处理
   * @param sparkSession
   */
  def execute(sparkSession: SparkSession)

  /**
   * 数据存储
   * dwd及dws层的数据都是需要写入到kudu数据库中，写入逻辑相同
   * @param dataFrame
   * @param isAutoCreateTable
   */
  def save(dataFrame:DataFrame, tableName:String, isAutoCreateTable:Boolean = true): Unit = {
    //允许自动创建表
    if (isAutoCreateTable) {
      Tools.autoCreateKuduTable(tableName, dataFrame)
    }

    //将数据写入到kudu中
    dataFrame.write.format(Configuration.SPARK_KUDU_FORMAT).options(Map(
      "kudu.master" -> Configuration.kuduRpcAddress,
      "kudu.table" -> tableName
    )).mode(SaveMode.Append).save()
  }
}