文章目录
- 一 DWM层-订单宽表
- 1 需求分析与思路
- 2 订单和订单明细关联代码实现
- （1）从Kafka的dwd层接收订单和订单明细数据
- a 创建订单实体类
- b 创建订单明细实体类
- c 在dwm包下创建OrderWideApp读取订单和订单明细数据
- d 测试
- （2）订单和订单明细关联（双流join）
- a 设定事件时间水位线
- b 创建合并后的宽表实体类
- c 设定关联的key
- d 订单和订单明细关联 intervalJoin
- e 测试
一 DWM层-订单宽表
1 需求分析与思路
订单是统计分析的重要的对象，围绕订单有很多的维度统计需求，比如用户、地区、商品、品类、品牌等等。
为了之后统计计算更加方便，减少大表之间的关联，所以在实时计算过程中将围绕订单的相关数据整合成为一张订单的宽表。
如上图，由于在之前的操作已经把数据分拆成了事实数据和维度数据，事实数据（绿色）进入kafka数据流（DWD层）中，维度数据（蓝色）进入hbase中长期保存。那么在DWM层中要把实时和维度数据进行整合关联在一起，形成宽表。那么这里就要处理有两种关联，事实数据和事实数据关联、事实数据和维度数据关联。
- 事实数据和事实数据关联，其实就是流与流之间的关联。
- 事实数据与维度数据关联，其实就是流计算中查询外部数据源。
2 订单和订单明细关联代码实现
（1）从Kafka的dwd层接收订单和订单明细数据
实现思路如下：
a 创建订单实体类
package com.hzy.gmall.realtime.beans;
import lombok.Data;
import java.math.BigDecimal;
/**
* Desc: 订单实体类
*/
@Data
public class OrderInfo {
// 属性名需要与数据库中字段名相同
Long id;
Long province_id;
String order_status;
Long user_id;
BigDecimal total_amount; //实际付款金额
BigDecimal activity_reduce_amount;
BigDecimal coupon_reduce_amount;
BigDecimal original_total_amount;
BigDecimal feight_fee;
String expire_time;
String create_time;
String operate_time;
String create_date; // 把其他字段处理得到
String create_hour;
Long create_ts; // 通过create_time转换
}
b 创建订单明细实体类
package com.hzy.gmall.realtime.beans;
import lombok.Data;
import java.math.BigDecimal;
/**
* Desc:订单明细实体类
*/
@Data
public class OrderDetail {
Long id;
Long order_id ; // 无外键约束
Long sku_id;
BigDecimal order_price ;
Long sku_num ;
String sku_name; // 冗余字段，可以减少连接查询的次数
String create_time;
BigDecimal split_total_amount;
BigDecimal split_activity_amount;
BigDecimal split_coupon_amount;
Long create_ts; // 通过create_time转换
}
c 在dwm包下创建OrderWideApp读取订单和订单明细数据
package com.hzy.gmall.realtime.app.dwm;
/**
* 订单宽表的准备
*/
public class OrderWideApp {
public static void main(String[] args) throws Exception {
//TODO 1 基本环境准备
//1.1 流处理环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//1.2 设置并行度
env.setParallelism(4);
//TODO 2 检查点设置（略）
//TODO 3 从kafka中读取数据
//3.1 声明消费主题以及消费者组
String orderInfoSourceTopic = "dwd_order_info";
String orderDetailSourceTopic = "dwd_order_detail";
String groupId = "order_wide_app_group";
//3.2 获取kafka消费者对象
// 订单
FlinkKafkaConsumer<String> orderInfoKafkaSource = MyKafkaUtil.getKafkaSource(orderInfoSourceTopic, groupId);
// 订单明细
FlinkKafkaConsumer<String> orderDetailKafkaSource = MyKafkaUtil.getKafkaSource(orderDetailSourceTopic, groupId);
//3.3 读取数据，封装为流
// 订单流
DataStreamSource<String> orderInfoStrDS = env.addSource(orderInfoKafkaSource);
// 订单明细流
DataStreamSource<String> orderDetailStrDS = env.addSource(orderDetailKafkaSource);
//TODO 4 对流中数据类型进行转换 String -> 实体对象
//订单
SingleOutputStreamOperator<OrderInfo> orderInfoDS = orderInfoStrDS.map(
new RichMapFunction<String, OrderInfo>() {
private SimpleDateFormat sdf;
@Override
public void open(Configuration parameters) throws Exception {
sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
}
@Override
public OrderInfo map(String jsonStr) throws Exception {
OrderInfo orderInfo = JSON.parseObject(jsonStr, OrderInfo.class);
orderInfo.setCreate_ts(sdf.parse(orderInfo.getCreate_time()).getTime());
return orderInfo;
}
}
);
// 订单明细
SingleOutputStreamOperator<OrderDetail> orderDetailDS = orderDetailStrDS.map(
new RichMapFunction<String, OrderDetail>() {
private SimpleDateFormat sdf;
@Override
public void open(Configuration parameters) throws Exception {
sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
}
@Override
public OrderDetail map(String jsonStr) throws Exception {
OrderDetail orderDetail = JSON.parseObject(jsonStr, OrderDetail.class);
orderDetail.setCreate_ts(sdf.parse(orderDetail.getCreate_time()).getTime());
return orderDetail;
}
}
);
orderInfoDS.print("订单信息：");
orderDetailDS.print("订单明细：");
env.execute();
}
}
d 测试
启动zookeeper、kafka、maxwell、hdfs，等待退出安全模式，启动Hbase
在配置表table_process中添加两条数据，如下
source_table operate_type sink_type sink_table sink_columns sink_pk sink_extend
order_detail insert kafka dwd_order_detail id,order_id,sku_id,sku_name,img_url,order_price,sku_num,create_time,source_type,source_id,split_total_amount,split_activity_amount,split_coupon_amount id (NULL)
order_info insert kafka dwd_order_info id,consignee,consignee_tel,total_amount,order_status,user_id,payment_way,delivery_address,province_id,activity_reduce_amount,coupon_reduce_amount,original_total_amount,feight_fee,feight_fee_reduce,refundable_time id (NULL)
启动BaseDBApp、OrderWideApp，模拟生成业务数据，观察结果。
- 执行流程
业务数据生成**->Maxwell同步->Kafka的ods_base_db_m主题->BaseDBApp分流写回kafka->dwd_order_info和dwd_order_detail->**OrderWideApp从kafka的dwd层读数据，打印输出
（2）订单和订单明细关联（双流join）
在flink中的流join大体分为两种，一种是基于时间窗口的join（Time Windowed Join），比如join、coGroup等。另一种是基于状态缓存的join（Temporal Table Join），比如intervalJoin。
这里选用intervalJoin，因为相比较窗口join，intervalJoin使用更简单，而且避免了应匹配的数据处于不同窗口的问题。intervalJoin目前只有一个问题，就是还不支持left join。
但是这里订单主表与订单从表之间的关联不需要left join，所以intervalJoin是较好的选择。
详情见官网关于窗口的说明。
Interval join 组合元素的条件为：两个流（我们暂时称为 A 和 B）中 key 相同且 B 中元素的 timestamp 处于 A 中元素 timestamp 的一定范围内。
这个条件可以更加正式地表示为 b.timestamp ∈ [a.timestamp + lowerBound; a.timestamp + upperBound] 或 a.timestamp + lowerBound <= b.timestamp <= a.timestamp + upperBound
这里的 a 和 b 为 A 和 B 中共享相同 key 的元素。上界和下界可正可负，只要下界永远小于等于上界即可。 Interval join 目前仅执行 inner join。
当一对元素被传递给 ProcessJoinFunction，他们的 timestamp 会从两个元素的 timestamp 中取最大值 （timestamp 可以通过 ProcessJoinFunction.Context 访问）。
Interval join 目前仅支持 event time。
intervalJoin连接数据的方式如下图：
上例中，我们 join 了橙色和绿色两个流，join 的条件是：以 -2 毫秒为下界、+1 毫秒为上界。 默认情况下，上下界也被包括在区间内，但 .lowerBoundExclusive() 和 .upperBoundExclusive() 可以将它们排除在外。
图中三角形所表示的条件也可以写成更加正式的表达式：
orangeElem.ts + lowerBound <= greenElem.ts <= orangeElem.ts + upperBound
a 设定事件时间水位线
// TODO 5 指定Watermark并提取事件时间字段
//订单
SingleOutputStreamOperator<OrderInfo> orderInfoWithWatermarkDS = orderInfoDS.assignTimestampsAndWatermarks(
WatermarkStrategy.<OrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(3))
.withTimestampAssigner(
new SerializableTimestampAssigner<OrderInfo>() {
@Override
public long extractTimestamp(OrderInfo orderInfo, long recordTimestamp) {
return orderInfo.getCreate_ts();
}
}
)
);
// 订单明细
SingleOutputStreamOperator<OrderDetail> orderDetailWithWatermarkDS = orderDetailDS.assignTimestampsAndWatermarks(
WatermarkStrategy.<OrderDetail>forBoundedOutOfOrderness(Duration.ofSeconds(3))
.withTimestampAssigner(
new SerializableTimestampAssigner<OrderDetail>() {
@Override
public long extractTimestamp(OrderDetail orderDetail, long recordTimestamp) {
return orderDetail.getCreate_ts();
}
}
)
);
b 创建合并后的宽表实体类
package com.hzy.gmall.realtime.beans;
import lombok.AllArgsConstructor;
import lombok.Data;
import org.apache.commons.lang3.ObjectUtils;
import java.math.BigDecimal;
/**
* Desc: 订单和订单明细关联宽表对应实体类
*/
@Data
@AllArgsConstructor
public class OrderWide {
Long detail_id;
Long order_id ;
Long sku_id;
BigDecimal order_price ;
Long sku_num ;
String sku_name;
Long province_id;
String order_status;
Long user_id;
BigDecimal total_amount;
BigDecimal activity_reduce_amount;
BigDecimal coupon_reduce_amount;
BigDecimal original_total_amount;
BigDecimal feight_fee;
BigDecimal split_feight_fee;
BigDecimal split_activity_amount;
BigDecimal split_coupon_amount;
BigDecimal split_total_amount;
String expire_time;
String create_time;
String operate_time;
String create_date; // 把其他字段处理得到
String create_hour;
String province_name;//查询维表得到
String province_area_code;
String province_iso_code;
String province_3166_2_code;
Integer user_age ;
String user_gender;
Long spu_id; //作为维度数据 要关联进来
Long tm_id;
Long category3_id;
String spu_name;
String tm_name;
String category3_name;
public OrderWide(OrderInfo orderInfo, OrderDetail orderDetail){
mergeOrderInfo(orderInfo);
mergeOrderDetail(orderDetail);
}
// 将订单的信息赋值给订单宽表
public void mergeOrderInfo(OrderInfo orderInfo ) {
if (orderInfo != null) {
this.order_id = orderInfo.id;
this.order_status = orderInfo.order_status;
this.create_time = orderInfo.create_time;
this.create_date = orderInfo.create_date;
this.activity_reduce_amount = orderInfo.activity_reduce_amount;
this.coupon_reduce_amount = orderInfo.coupon_reduce_amount;
this.original_total_amount = orderInfo.original_total_amount;
this.feight_fee = orderInfo.feight_fee;
this.total_amount = orderInfo.total_amount;
this.province_id = orderInfo.province_id;
this.user_id = orderInfo.user_id;
}
}
// 将订单明细的信息赋值给订单宽表
public void mergeOrderDetail(OrderDetail orderDetail ) {
if (orderDetail != null) {
this.detail_id = orderDetail.id;
this.sku_id = orderDetail.sku_id;
this.sku_name = orderDetail.sku_name;
this.order_price = orderDetail.order_price;
this.sku_num = orderDetail.sku_num;
this.split_activity_amount=orderDetail.split_activity_amount;
this.split_coupon_amount=orderDetail.split_coupon_amount;
this.split_total_amount=orderDetail.split_total_amount;
}
}
// firstNonNull获取参数中第一个不为空的值
public void mergeOtherOrderWide(OrderWide otherOrderWide){
this.order_status = ObjectUtils.firstNonNull( this.order_status ,otherOrderWide.order_status);
this.create_time = ObjectUtils.firstNonNull(this.create_time,otherOrderWide.create_time);
this.create_date = ObjectUtils.firstNonNull(this.create_date,otherOrderWide.create_date);
this.coupon_reduce_amount = ObjectUtils.firstNonNull(this.coupon_reduce_amount,otherOrderWide.coupon_reduce_amount);
this.activity_reduce_amount = ObjectUtils.firstNonNull(this.activity_reduce_amount,otherOrderWide.activity_reduce_amount);
this.original_total_amount = ObjectUtils.firstNonNull(this.original_total_amount,otherOrderWide.original_total_amount);
this.feight_fee = ObjectUtils.firstNonNull( this.feight_fee,otherOrderWide.feight_fee);
this.total_amount = ObjectUtils.firstNonNull( this.total_amount,otherOrderWide.total_amount);
this.user_id = ObjectUtils.<Long>firstNonNull(this.user_id,otherOrderWide.user_id);
this.sku_id = ObjectUtils.firstNonNull( this.sku_id,otherOrderWide.sku_id);
this.sku_name = ObjectUtils.firstNonNull(this.sku_name,otherOrderWide.sku_name);
this.order_price = ObjectUtils.firstNonNull(this.order_price,otherOrderWide.order_price);
this.sku_num = ObjectUtils.firstNonNull( this.sku_num,otherOrderWide.sku_num);
this.split_activity_amount=ObjectUtils.firstNonNull(this.split_activity_amount);
this.split_coupon_amount=ObjectUtils.firstNonNull(this.split_coupon_amount);
this.split_total_amount=ObjectUtils.firstNonNull(this.split_total_amount);
}
}
c 设定关联的key
// TODO 6 通过分组指定两流的关联字段 -- order_id
// 订单
KeyedStream<OrderInfo, Long> orderInfoKeyedDS = orderInfoWithWatermarkDS.keyBy(OrderInfo::getId);
// 订单明细
KeyedStream<OrderDetail, Long> orderDetailkeyedDS = orderDetailWithWatermarkDS.keyBy(OrderDetail::getOrder_id);
d 订单和订单明细关联 intervalJoin
这里设置了正负5秒，以防止在业务系统中主表与从表保存的时间差。
// TODO 7 双流join，使用intervalJoin
// 用订单（一）join订单明细（多）
SingleOutputStreamOperator<OrderWide> orderWideDS = orderInfoKeyedDS
.intervalJoin(orderDetailkeyedDS)
.between(Time.seconds(-5), Time.seconds(5))
.process(
new ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>() {
@Override
public void processElement(OrderInfo orderInfo, OrderDetail orderDetail, Context ctx, Collector<OrderWide> out) throws Exception {
out.collect(new OrderWide(orderInfo, orderDetail));
}
}
);
orderWideDS.print(">>>");
e 测试
目前以完成工作
-
基本环境准备
-
设置检查点
-
从kafka中读取两条流数据并在转换结构的时候补充创建时间（create_ts）
-
通过keyby设置关联字段 – order_id
-
双流join
A.intervalJoin(B) .between(下界,上界) .process() -
测试，同2（1）d
