支付宽表

支付宽表的目的，最主要的原因是支付表没有到订单明细，支付金额没有细分到商品上，没有办法统计商品级的支付状况。所以本次宽表的核心就是要把支付表的信息与订单明细关联上。

解决方案有两个

一个是把订单明细表（或者宽表）输出到 Hbase 上，在支付宽表计算时查询 hbase，这相当于把订单明细作为一种维度进行管理。

一个是用流的方式接收订单明细，然后用双流 join 方式进行合并。因为订单与支付产生有一定的时差。所以必须用 intervalJoin 来管理流的状态时间，保证当支付到达时订单明细还保存在状态中。

支付相关实体类

PaymentInfo.java：支付实体类

import lombok.Data;
import java.math.BigDecimal;
/*** @author zhangbaohpu* @date 2021/12/25 10:08* @desc 支付实体类*/
@Data
public class PaymentInfo {Long id;Long order_id;Long user_id;BigDecimal total_amount;String subject;String payment_type;String create_time;String callback_time;
}

PaymentWide.java：支付宽表实体类

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.commons.beanutils.BeanUtils;
import java.lang.reflect.InvocationTargetException;
import java.math.BigDecimal;
/*** @author zhangbaohpu* @date 2021/12/25 10:10* @desc 支付宽表实体类*/
@Data
@AllArgsConstructor
@NoArgsConstructor
public class PaymentWide {Long payment_id;String subject;String payment_type;String payment_create_time;String callback_time;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 order_create_time;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 PaymentWide(PaymentInfo paymentInfo, OrderWide orderWide){mergeOrderWide(orderWide);mergePaymentInfo(paymentInfo);}public void mergePaymentInfo(PaymentInfo paymentInfo ) {if (paymentInfo != null) {try {BeanUtils.copyProperties(this,paymentInfo);payment_create_time=paymentInfo.create_time;payment_id = paymentInfo.id;} catch (IllegalAccessException e) {e.printStackTrace();} catch (InvocationTargetException e) {e.printStackTrace();}}}public void mergeOrderWide(OrderWide orderWide ) {if (orderWide != null) {try {BeanUtils.copyProperties(this,orderWide);order_create_time=orderWide.create_time;} catch (IllegalAccessException e) {e.printStackTrace();} catch (InvocationTargetException e) {e.printStackTrace();}}}
}

支付宽表主程序

目前还没有任何计算，仍然放在dwm层

在dwm包下创建PaymentWideApp.java任务类

import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUnit;
import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson.JSON;
import com.zhangbao.gmall.realtime.bean.OrderWide;
import com.zhangbao.gmall.realtime.bean.PaymentInfo;
import com.zhangbao.gmall.realtime.bean.PaymentWide;
import com.zhangbao.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.util.Collector;import java.time.Duration;/*** @author zhangbaohpu* @date 2021/12/25 10:16* @desc 支付宽表*/
public class PaymentWideApp {public static void main(String[] args) {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//添加并行度env.setParallelism(4);//设置检查点
//        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setCheckpointTimeout(60000);
//        env.setStateBackend(new FsStateBackend("hdfs://hadoop101:9000/gmall/flink/checkpoint/paymentWide"));
//        //指定哪个用户读取hdfs文件
//        System.setProperty("HADOOP_USER_NAME","zhangbao");//设置kafka主题及消费者组String paymentInfoTopic = "dwd_payment_info";String orderWideTopic = "dwm_order_wide";String paymentWideTopic = "dwm_payment_wide";String paymentWideGroup = "paymentWideGroup";//获取支付信息FlinkKafkaConsumer<String> paymentInfo = MyKafkaUtil.getKafkaSource(paymentInfoTopic, paymentWideGroup);DataStreamSource<String> paymentInfoJsonStrDs = env.addSource(paymentInfo);//获取订单宽表信息FlinkKafkaConsumer<String> orderWide = MyKafkaUtil.getKafkaSource(orderWideTopic, paymentWideGroup);DataStreamSource<String> orderWideJsonStrDs = env.addSource(orderWide);//转换格式SingleOutputStreamOperator<PaymentInfo> paymentJsonDs = paymentInfoJsonStrDs.map(paymentInfoStr -> JSON.parseObject(paymentInfoStr, PaymentInfo.class));SingleOutputStreamOperator<OrderWide> orderWideJsonDs = orderWideJsonStrDs.map(orderWideStr -> JSON.parseObject(orderWideStr, OrderWide.class));paymentJsonDs.print("payment info >>>");orderWideJsonDs.print("order wide >>>");//指定事件时间字段SingleOutputStreamOperator<PaymentInfo> paymentInfoWithWaterMarkDs = paymentJsonDs.assignTimestampsAndWatermarks(WatermarkStrategy.<PaymentInfo>forBoundedOutOfOrderness(Duration.ofSeconds(3)).withTimestampAssigner(new SerializableTimestampAssigner<PaymentInfo>() {@Overridepublic long extractTimestamp(PaymentInfo paymentInfo, long l) {return DateUtil.parse(paymentInfo.getCallback_time(), DatePattern.NORM_DATETIME_PATTERN).getTime();}}));SingleOutputStreamOperator<OrderWide> orderWideWithWaterMarkDs = orderWideJsonDs.assignTimestampsAndWatermarks(WatermarkStrategy.<OrderWide>forBoundedOutOfOrderness(Duration.ofSeconds(3)).withTimestampAssigner(new SerializableTimestampAssigner<OrderWide>() {@Overridepublic long extractTimestamp(OrderWide orderWide, long l) {return DateUtil.parse(orderWide.getCreate_time(), DatePattern.NORM_DATETIME_PATTERN).getTime();}}));//分组KeyedStream<PaymentInfo, Long> paymentInfoKeyedDs = paymentInfoWithWaterMarkDs.keyBy(payInfoObj -> payInfoObj.getOrder_id());KeyedStream<OrderWide, Long> orderWideKeyedDs = orderWideWithWaterMarkDs.keyBy(orderWideObj -> orderWideObj.getOrder_id());paymentInfoKeyedDs.print("paymentInfoKeyedDs >>>");orderWideKeyedDs.print("orderWideKeyedDs >>>");//双流join，用支付数据关联订单数据SingleOutputStreamOperator<PaymentWide> paymentWideObjDs = paymentInfoKeyedDs.intervalJoin(orderWideKeyedDs).between(Time.seconds(-1800), Time.seconds(1800)).process(new ProcessJoinFunction<PaymentInfo, OrderWide, PaymentWide>() {@Overridepublic void processElement(PaymentInfo paymentInfo, OrderWide orderWide, ProcessJoinFunction<PaymentInfo, OrderWide, PaymentWide>.Context context, Collector<PaymentWide> collector) throws Exception {System.out.println(paymentInfo);System.out.println(orderWide);collector.collect(new PaymentWide(paymentInfo, orderWide));}});//将数据流转换为jsonSingleOutputStreamOperator<String> paymentWideDs = paymentWideObjDs.map(paymentWide -> JSON.toJSONString(paymentWide));paymentWideDs.print("payment wide json >>> ");//发送到kafkaFlinkKafkaProducer<String> kafkaSink = MyKafkaUtil.getKafkaSink(paymentWideTopic);paymentWideDs.addSink(kafkaSink);try {env.execute("payment wide task");} catch (Exception e) {e.printStackTrace();}}
}

到这里，支付宽表的操作就完成了。
项目地址：https://github.com/zhangbaohpu/gmall-flink-parent/tree/master/gmall-realtime

总结

DWM 层部分的代码主要的责任，是通过计算把一种明细转变为另一种明细以应对后续的统计。学完本阶段内容要求掌握

学会利用状态（state）进行去重操作。（需求：UV 计算）
学会利用 CEP 可以针对一组数据进行筛选判断。需求：跳出行为计算
学会使用 intervalJoin 处理流 join
学会处理维度关联，并通过缓存和异步查询对其进行性能优化。

更多请在某公号平台搜索：选手一号位，本文编号：1011，回复即可获取。

11.Flink实时项目之支付宽表相关推荐

9.Flink实时项目之订单宽表
1.需求分析订单是统计分析的重要的对象,围绕订单有很多的维度统计需求,比如用户.地区.商品.品类.品牌等等.为了之后统计计算更加方便,减少大表之间的关联,所以在实时计算过程中将围绕订单的相关数据整合 ...
4.Flink实时项目之数据拆分
Python微信订餐小程序课程视频 https://edu.csdn.net/course/detail/36074 Python实战量化交易理财系统 https://edu.csdn.net/cou ...
Flink实时销售宽表构建
背景:目前在一家电商公司,对报表的实时性要求很高.实时性要求较高的场景,比如: 1.集团各个分公司对商品配送过程中生成的各个单据的对账实时性很高. 2.采购部依赖商品的平均进价对客户进行报价,所以对商 ...
mysql 数据展示装置_实时生成数据宽表的方法和装置与流程
本发明涉及计算机技术领域,尤其涉及一种实时生成数据宽表的方法和装置. 背景技术: 数据仓库是面向主题的.集成的.相对稳定的.随时间不短变化得数据集合,用以支持经营管理中的决策制定.数据仓库中的数据面向 ...
Flink 实时计算 - 维表 Join 解读
Flink 实时计算 - 维表 Join 解读前言 Flink 1.9 版本可以说是一个具有里程碑意义的版本,其内部合入了很多 Blink Table/SQL 方面的功能,同时也开始增强 Flink ...
Flink入门(五) 实时流Join ElasticSearch6维度表
需求实时流需要和维护表Join做属性的扩展. Spark-Streaming可以 stream join hive表. flink没发现这个功能,所以将维度表放在ES上. maven依赖 <p ...
Flink实时数据处理实践经验(Flink去重、维表关联、定时器、双流join)
Flink实时数据处理实践经验文章目录 Flink实时数据处理实践经验 1. 数据输入与预处理 2. 实时数据处理 3. 实时数仓架构 4. 优化方案 Java.大数据开发学习要点(持续更新中-) ...
Flink cdc+ doris 大宽表实践~
还没整理好,别慌. 一,业务问题: 多个表关联join(涉及时间维度跨度很长),几乎等同于全量关联,这个时候flink sql join没法做,因为state会无线增大,然后OOM. 二,解决方案 : ...
Elasticseach：从微服务架构演变到大宽表思维的架构转变
序言图示:Elasticsearch 在DB-Engine综合排名第8 Elasticsearch 简称"ES", 在DB-Engine 综合排名第8,已经持续了相当长的时间,按 ...

11.Flink实时项目之支付宽表

支付宽表

支付相关实体类

支付宽表主程序

总结

11.Flink实时项目之支付宽表相关推荐

最新文章

热门文章