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

虚幻大学 xuhss 391℃ 0评论

Python微信订餐小程序课程视频

https://blog.csdn.net/m0_56069948/article/details/122285951

Python实战量化交易理财系统

https://blog.csdn.net/m0_56069948/article/details/122285941

支付宽表

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

解决方案有两个

一个是把订单明细表(或者宽表)输出到 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 paymentInfo = MyKafkaUtil.getKafkaSource(paymentInfoTopic, paymentWideGroup);
        DataStreamSource paymentInfoJsonStrDs = env.addSource(paymentInfo);
        //获取订单宽表信息
        FlinkKafkaConsumer orderWide = MyKafkaUtil.getKafkaSource(orderWideTopic, paymentWideGroup);
        DataStreamSource orderWideJsonStrDs = env.addSource(orderWide);
​
        //转换格式
        SingleOutputStreamOperator paymentJsonDs = paymentInfoJsonStrDs.map(paymentInfoStr -> JSON.parseObject(paymentInfoStr, PaymentInfo.class));
        SingleOutputStreamOperator orderWideJsonDs = orderWideJsonStrDs.map(orderWideStr -> JSON.parseObject(orderWideStr, OrderWide.class));
​
        paymentJsonDs.print("payment info >>>");
        orderWideJsonDs.print("order wide >>>");
​
        //指定事件时间字段
        SingleOutputStreamOperator paymentInfoWithWaterMarkDs = paymentJsonDs.assignTimestampsAndWatermarks(
            WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(3))
               .withTimestampAssigner(new SerializableTimestampAssigner() {
                    @Override
                    public long extractTimestamp(PaymentInfo paymentInfo, long l) {
                        return DateUtil.parse(paymentInfo.getCallback\_time(), DatePattern.NORM\_DATETIME\_PATTERN).getTime();
                   }
               })
       );
        SingleOutputStreamOperator orderWideWithWaterMarkDs = orderWideJsonDs.assignTimestampsAndWatermarks(
            WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(3))
               .withTimestampAssigner(new SerializableTimestampAssigner() {
                    @Override
                    public long extractTimestamp(OrderWide orderWide, long l) {
                        return DateUtil.parse(orderWide.getCreate\_time(), DatePattern.NORM\_DATETIME\_PATTERN).getTime();
                   }
               })
       );
​
        //分组
        KeyedStream paymentInfoKeyedDs = paymentInfoWithWaterMarkDs.keyBy(payInfoObj -> payInfoObj.getOrder\_id());
        KeyedStream orderWideKeyedDs = orderWideWithWaterMarkDs.keyBy(orderWideObj -> orderWideObj.getOrder\_id());
​
        paymentInfoKeyedDs.print("paymentInfoKeyedDs >>>");
        orderWideKeyedDs.print("orderWideKeyedDs >>>");
​
        //双流join,用支付数据关联订单数据
        SingleOutputStreamOperator paymentWideObjDs = paymentInfoKeyedDs.intervalJoin(orderWideKeyedDs)
               .between(Time.seconds(-1800), Time.seconds(1800))
               .process(new ProcessJoinFunction() {
                    @Override
                    public void processElement(PaymentInfo paymentInfo, OrderWide orderWide, ProcessJoinFunction.Context context, Collector collector) throws Exception {
                        System.out.println(paymentInfo);
                        System.out.println(orderWide);
                        collector.collect(new PaymentWide(paymentInfo, orderWide));
                   }
               });
        //将数据流转换为json
        SingleOutputStreamOperator paymentWideDs = paymentWideObjDs.map(paymentWide -> JSON.toJSONString(paymentWide));
        paymentWideDs.print("payment wide json >>> ");
        //发送到kafka
        FlinkKafkaProducer 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,回复即可获取。

转载请注明:xuhss » 11.Flink实时项目之支付宽表

喜欢 (0)

您必须 登录 才能发表评论!