Flink实时电商数仓(十)

common模块回顾

  1. app
    • BaseApp: 作为其他子模块中使用Flink - StreamAPI的父类,实现了StreamAPI中的通用逻辑,在其他子模块中只需编写关于数据处理的核心逻辑。
    • BaseSQLApp: 作为其他子模块中使用Flink- SQLAPI的父类。在里面设置了使用SQL API的环境、并行度、检查点等固定逻辑。
  2. bean:存放其他子模块中使用到的javaBean对象,因为如果一直使用jsonObject对象调用数据的话,需要使用类似getString("字段名")的方式,没有直接使用javaBean对象那么方便。
  3. constant
    • 存储字符串常量
    • 为了保证一致性,如果某个常量修改时,只需在这里修改即可对整个项目进行修改
  4. function
    • DorisMapFunction:将javaBean对象转换为对应的json字符串对象,并且将驼峰式命名方式修改为蛇形命名方式。便于写入doris。
  5. util
    • DateFormateUtil
    • FlinkSinkUtil
    • FlinkSourceUtil
    • HBaseUtil
    • IkUtil
    • JdbcUtil
    • SQLUtil
      • getUpsertKafakaSQL: 一定要声明主键,支持撤回流
      • getDorisSinkSQL: 用于写入Doris

dim层回顾

  • Flink-cdc监控mysql中的维度配置表
  • 将监控的数据流做成广播流
  • 将广播流和读取数据的主流进行connect
  • 主流数据根据广播流的配置信息进行分流,注意需要先提前缓存一次配置表信息
  • 达到动态拆分数据表的效果

dwd层FlinkSQL回顾

  • 注意join时会将所有数据都存储到内存中,需要考虑设置TTL
  • 大表join小表时,可以考虑使用lookup join
  • 如果数据流有明确的先后关系时,考虑使用Interval join

在支付成功模块,由于订单详情表处理时已经存在撤回流,但支付成功模块也是使用left join方式调用订单详情数据,会导致产生两次撤回流。在后续dws层处理时,要注意对数据进行去重过滤。

dws层回顾

  • 如何判断使用FlinkSQL还是StreamAPI
    • 如果比较标准化, 比如简单的开窗聚合,一般使用FlinkSQL
    • 如果需要使用状态处理数据,比如判断是否为独立用户,使用StreamAPI

交易域sku粒度订单下单各窗口汇总

  • 需求分析:从Kafka订单明细主题读取数据,过滤null数据并按照唯一键对数据去重,按照SKU维度分组,统计原始金额、活动减免金额、优惠券减免金额和订单金额,并关联维度信息,将数据写入Doris交易域SKU粒度下单各窗口汇总表

  • 思路分析:

    • 方案一:按照订单ID进行分组,根据业务逻辑设置定时器取最后一个数据进行发送
    • 方案二:将度量值存放到状态中,每次新数据到达时,将新的度量值减去状态中的度量值
  • 具体实现

    • 因为需要使用状态,故使用BaseApp; 设置端口号10029,并发度4,消费者组为类名,消费者主题名称为dwd订单详情
    • 读取dwd下单主题数据, stream.print()
    • 过滤清洗:
      • 去掉null数据, stream.flatMap(new FlatMapFunction<>())
      • ts: 水位线,不能为空;进行位数的修正,如果是10位的,使用 jsonObj.put("ts", ts*1000)
      • id: keyby的关键字,不能为空
      • sku_id: group by的粒度关键字,也不能为空
    • 添加水位线
      • 网络延迟5L
      • 添加数据的泛型,提取数据中的ts,作为水位线(注意观察ts的位数,需要为13位,毫秒级)
    • 修正度量值,转换数据结构
      • 使用id关键字进行分组
      • 使用process算子中的状态来进行处理stream.process(new KeyedProcessFunction<>),返回值为对应的javabean对象
      • 在状态中存储上一次的度量值大小,只保存30秒
      • processElement()方法中获取状态中的度量值,使用前需要判空,如果为空设置为0,之后才能进行数值计算。
      • 创建对应的bean对象,度量值都减去状态中的度量值和更新状态中当前的度量值
    • 分组开窗聚合
      • 使用skuId进行keyby
      • 分组后使用window算子进行开窗,设置窗口时间,注意Time属于org.apache.flink.streaming.api.windowing.time.Time.seconds()
      • 使用reduce算子进行聚合计算, 聚合时需要累积所有度量值
      • new ProcessWindowFunction()获取窗口信息, startTime, EndTime, curTime, 获取到后写入javaBean对象中
    • 关联维度信息
      • 先分组聚合再关联维度信息的原因:关联维度信息需要join操作,是很耗费性能的大操作。先聚合数据能大幅度减少数据量。
      • 启动HBase,查看对的sku_info表中是否存储着对应的维度信息
      • 获取外部连接,需要使用生命周期方法(open,close在整个算子执行过程中只运行一次);对应的关联维度信息,即RichMapFunction()
      • map方法中使用HBase的API读取表格数据,使用读取到的字段补全原本的信息
    • 创建HBase的API:读取表格数据 get
      • 获取table
      • 创建get对象
      • 调用get方法
      • 获取数据写入jsonObj
    • 写出到Doris

维度关联优化

  1. 旁路缓存:独立缓存服务有(redis, memcache).
    在这里插入图片描述
  • 使用旁路缓存时要注意保持数据的一致性,如果数据发生修改和删除,直接删除redis中的数据。

同步+旁路缓存模式

  1. 引入Jedis相关依赖
<dependency>
    <groupId>redis.clients</groupId>
    <artifactId>jedis</artifactId>
</dependency>
  1. 创建Redis工具类RedisUtil
  2. 在RichMapFunction中的open和close方法中获取和关闭HBase和Redisd的连接。
  3. 拼接对应的redisRowKey
  4. 读取Redis缓存的数据,jsonObj的字符串
  5. 判断redis读取到的数据是否为空
    • 没有数据:需要读取HBase;jsonObj = HBaseUtil.getCells(), 读取到数据后,使用jedis.setex()存储到redis
    • redis有缓存,直接返回
  6. 进行维度关联

Dim层写入HBase修正

  • 在dim层将数据写入HBase时,需要同时获取Redis的连接。
  • 判断redis中的缓存是否发生变化
    • 判断数据类型是修改或删除时,删除Redis中对应的数据
    • 拼写数据的rowkey
    • 使用jedis.del(rediskey)来删除