主题及指标开发
一、主题开发业务流程
二、离线模块初始化
1、创建包结构
2、创建时间处理工具
3、定义主题宽表及指标结果表的表名
4、物流字典码表数据类型定义枚举类
5、封装公共接口
主题及指标开发
一、主题开发业务流程
二、离线模块初始化
1、创建包结构
本次项目采用scala编程语言，因此创建scala目录
|
包名 |
说明 |
|---|---|
|
cn.it.logistics.offline |
离线指标统计程序所在包 |
|
cn.it.logistics.offline.dwd |
离线指标dwd层程序所在包 |
|
cn.it.logistics.offline.dws |
离线指标dws层程序所在包 |
2、创建时间处理工具
实现步骤：
- 在公共模块的scala目录下的common程序包下创建DateHelper对象
- 实现获取当前日期
- 实现获取昨天日期
package cn.it.logistics.commonimport 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表
|
name |
type |
|---|---|
|
注册渠道 |
1 |
|
揽件状态 |
2 |
|
派件状态 |
3 |
|
快递员状态 |
4 |
|
地址类型 |
5 |
|
网点状态 |
6 |
|
员工状态 |
7 |
|
是否保价 |
8 |
|
运输工具类型 |
9 |
|
运输工具状态 |
10 |
|
仓库类型 |
11 |
|
是否租赁 |
12 |
|
货架状态 |
13 |
|
回执单状态 |
14 |
|
出入库类型 |
15 |
|
客户类型 |
16 |
|
下单终端类型 |
17 |
|
下单渠道类型 |
18 |
实现步骤：
- 在公共模块的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.offlineimport 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()}
}
