实时灵活分析需求（双流join）

墨蓝 2022-11-02 13:22 127阅读 0赞

### 目录 ###

*  一、实时灵活分析需求
 *   *  1.需求分析的结果
     *  2.实时场景的关键
 *  二、实现过程
 *   *  1.使用canal实时采集数据
     *  2.Join的过程
     *  3.双流Join
     *  4.redis的相关建模
 *  三、代码开发

# 一、实时灵活分析需求 #

## 1.需求分析的结果 ##

涉及全文检索，需要使用ES存储数据！

搜索的是商品明细，需要将商品的明细导入到ES！

商品明细：商品明细，男女比例，年龄比例

从Mysql的业务数据中取数据！

数据源： Mysql的业务数据！为了得到明细数据，需要三张表！

select 
    
    	order_detail中的详情， age, gender
    
    from
    
     order_detail  od  
    
    left join   order_info oi  on  od.order_id = oi.id
    
    left join  user_info u on oi.user_id=u.id

## 2.实时场景的关键 ##

问题：

由于网络的延迟，可能会造成两个流要Join的数据分别位于两个批次，完美错过！

分别位于两个批次，要么批次早到，要么批次迟到！

# 二、实现过程 #

## 1.使用canal实时采集数据 ##

order\_info: 新增和变化

canal采集时，只是为了获取order\_info中的user\_id，user\_id在订单创建时生成，无法被修改！

canal监控时，可以只监控 新增的数据！

order\_detail : 新增

user\_info : 新增和变化

## 2.Join的过程 ##

order\_info 和order\_detail 几乎是同时产生，可以直接进行Join!

order\_info 下单的用户，可能是之前产生的，当前canal监控不到，kafka中可能没有当前的用户信息！因此join用户时，需要Join，全量的用户表！

可以从mysql中读取全量的用户表！

可以为了性能，使用canal将所有新增的用户都写入到redis中，也需要保证redis中有全量的用户！从redis中读取用户信息，再和之前的order\_info和order\_detail join！

## 3.双流Join ##

问题：

由于网络的延迟，可能会造成两个流要Join的数据分别位于两个批次，完美错过！

分别位于两个批次，要么批次早到，要么批次迟到！

解决：将由于 早来 或 晚来，导致无法在当前批次 Join上的的数据，进行持久化！例如保证性能，将这部分数据，持久化到redis缓存！

实现过程： ①两个流当前批次的数据先进行Join，能匹配上就匹配上

②当前批次无法Join 成功的 order\_detail ，到redis中读取 早到的 order\_info 进行 Join

如果还Join 不成功， 说明 order\_detail 早来了！

此时，将order\_detail 写入redis，供后续的order\_info 进行 Join

③每个批次在Join时 ， order\_info 都需要从redis中查询，是否有早到的 order\_detail，如果有就关联

④由于order\_info 和 order-detail 是 1对N的关系，所以每个批次的order\_info 都需要无条件地写入redis, 等候后续迟到的order\_detail

因为order\_info 和 order\_detail都需要缓存到redis，可以根据公司集群的最大延迟，配置key的过期时间！即便不配置也可以，redis可以根据默认LRU的过期策略，清理冷门的key!

⑤使用full outer join 来判断哪些join不上，join不上的会为None

## 4.redis的相关建模 ##

order\_info 都需要无条件地写入redis：

单值：　string ,hash

集合：　　list, set ,zset

Ｋ： 不能是集合类型，因为已经组合成功的　order\_info　有自己的过期时间！而不是所有的order\_info放入一个集合中，使用同一个过期时间！

从集合中取出后，需要遍历集合才能和order\_detail匹配，效率低！

原则：　每个order\_info　都　自己唯一的key

order\_info:orderid

Ｖ：　 String

--------------------

早到的order\_detail 写入redis:

一个order\_id 对应 多个 order\_detail

K: order\_id标识

order\_detail:orderid

V: Set

--------------------

用户的维度信息：

一个用户保存一个k-v对

K： user:id

V： 用户的信息以json格式写入

string

# 三、代码开发 #

双流join处理

object SaleDetailApp { 
      
      def main(args: Array[String]): Unit = { 
        
        val streamingContext: StreamingContext = new StreamingContext("local[*]","SaleDetailApp",Seconds(10))
        
        //从kafka获取两个DS
        val ds1: InputDStream[ConsumerRecord[String, String]] = MykafkaUtil.getKafkaStream(GmallConstants.KAFKA_TOPIC_NEW_ORDER, streamingContext)
        val ds2: InputDStream[ConsumerRecord[String, String]] = MykafkaUtil.getKafkaStream(GmallConstants.KAFKA_TOPIC_ORDER_DETAIL,streamingContext)
        
        //从kafka的record中取出数据封装为样例类
        val orderInfoDs: DStream[(String, OrderInfo)] = ds1.map(recoord => { 
          val orderInfo: OrderInfo = JSON.parseObject(recoord.value(), classOf[OrderInfo])
          val format1: SimpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
          val format2: SimpleDateFormat = new SimpleDateFormat("HH")
          val format3: SimpleDateFormat = new SimpleDateFormat("yyyy-MM-dd")
        
          val date: Date = format1.parse(orderInfo.create_time)
          orderInfo.create_hour = format2.format(date)
          orderInfo.create_date = format3.format(date)
        
          //对敏感数据脱敏
          orderInfo.consignee_tel = orderInfo.consignee_tel.replaceAll("(\\w{3})\\w*(\\w{4})", "$1****$2")
        
          (orderInfo.id, orderInfo)
        })
      
        val orderDetailDs: DStream[(String, OrderDetail)] = ds2.map(record => { 
          val orderDetail: OrderDetail = JSON.parseObject(record.value(), classOf[OrderDetail])
          (orderDetail.order_id, orderDetail)
        })
        
        
        //① 两个流 full join
        val joinedDS: DStream[(String, (Option[OrderInfo], Option[OrderDetail]))] = orderInfoDs.fullOuterJoin(orderDetailDs)
        
        //以分区为单位操作，有返回值
        val noUserInfoDS: DStream[SaleDetail] = joinedDS.mapPartitions(iter => { 
          //必须放入算子中，因为无法序列化
          implicit var format = org.json4s.DefaultFormats
        
          //一个分区只创建一个redis连接
          val jedisClient: Jedis = RedisUtil.getJedisClient
        
          val saleDetails = ListBuffer[SaleDetail]()
        
          iter.foreach { 
            case (orderId, (orderInfoOption, orderDetailOption)) => { 
            
              val order_infoKey = "order_info:" + orderId
              val order_detailkey = "order_detail:" + orderId
            
              //①full join 要么order_info不为none，要么order_info为none，先处理join后的order_info
              //a)判断每个order_info join 的order_deail是否为none，不为none，直接组装成SaleDetail对象
              if (orderInfoOption != None) { 
                val orderInfo: OrderInfo = orderInfoOption.get
              
                if (orderDetailOption != None) { 
                  val orderDetail: OrderDetail = orderDetailOption.get
                  //封装组合数据
                  val saleDetail1: SaleDetail = new SaleDetail(orderInfo, orderDetail)
                
                  saleDetails.append(saleDetail1)
                }
              
                //b) orderInfo无条件写入redis
                val orderInfoJson: String = Serialization.write(orderInfo)
              
                jedisClient.setex(order_infoKey, 500, orderInfoJson)
              
                //c)从redis中查询，是否有早到的order_detail,如果有，就查出来，组装为SaleDetail
                val orderDetails: util.Set[String] = jedisClient.smembers(order_detailkey)
              
                //将java集合转换为scala集合
                orderDetails.asScala.foreach(orderDetailsStr => { 
                  val orderDetail: OrderDetail = JSON.parseObject(orderDetailsStr, classOf[OrderDetail])
                
                  val saleDetail: SaleDetail = new SaleDetail(orderInfo, orderDetail)
                
                  saleDetails.append(saleDetail)
                })
              
              } else { 
              
                //③再处理 join后的order_detail
                //order_detail 一定不是none
                val orderDetail: OrderDetail = orderDetailOption.get
              
                //e)取出order_detail
                //在redis中查看是否有早到的order_info,匹配上就组装为SaleDetail
              
                val orderInfoStr: String = jedisClient.get(order_infoKey)
              
                if (orderInfoStr != null) { 
                
                  val orderInfo: OrderInfo = JSON.parseObject(orderInfoStr, classOf[OrderInfo])
                
                  val saleDetail: SaleDetail = new SaleDetail(orderInfo, orderDetail)
                
                  saleDetails.append(saleDetail)
                } else { 
                  //没找到，意味着order_detail早到了，存入redis
                  val orderDetailJson: String = Serialization.write(orderDetail)
                
                  jedisClient.sadd(order_detailkey, orderDetailJson)
                  jedisClient.expire(order_detailkey, 500)
                }
              }
            }
          }
          //归还连接
          jedisClient.close()
        
          saleDetails.iterator
        
        })
        
        
        //读取redis中用户信息，进行关联
        val result: DStream[SaleDetail] = noUserInfoDS.mapPartitions(iter => { 
        
          //创建redis客户端
          val jedisClient: Jedis = RedisUtil.getJedisClient
        
          val saleDetails: Iterator[SaleDetail] = iter.map(saleDetail => { 
            //生成key
            val userKey = "user:" + saleDetail.user_id
          
            val userInfoStr: String = jedisClient.get(userKey)
      
            // userInfo数据 Json串在解析还原为 Object时，可以会由于Json串少 { 报错
            //Caused by: com.alibaba.fastjson.JSONException: syntax error, expect {, actual error, pos 0, fastjson-version
           // val userInfo: UserInfo = JSON.parseObject(userInfoStr, classOf[UserInfo])
          
                   val gson: Gson = new Gson()
                    val userInfo: UserInfo = gson.fromJson(userInfoStr, classOf[UserInfo])
          
            saleDetail.mergeUserInfo(userInfo)
          
            saleDetail
          
          })
          saleDetails
        
          jedisClient.close()
        
          saleDetails
        
        })
        
        result.print()
        
        //将最终结果写入ES
        result.foreachRDD(rdd=>{ 
    
          rdd.foreachPartition(iter=>{ 
    
            //每天的明细数据保存在一个index中
            val indexName: String = GmallConstants.ES_INDEX_SALE_DETAIL + LocalDate.now()
    
            //每条数据的id设计：order_detail 的 id
            val datas: List[(String, SaleDetail)] = iter.toList.map(saleDetail => { 
              (saleDetail.order_detail_id, saleDetail)
            })
    
            MyESUtil.insertBulk(indexName, datas)
    
          })
    
        })
      
      
        streamingContext.start()
      
        streamingContext.awaitTermination()
      
      
      }
      
      
    }

用户业务数据写入redis

package com.gmall.app
    
    import com.alibaba.fastjson.JSON
    import com.gmall.common.constansts.GmallConstants
    import com.gmall.util.{ MykafkaUtil, RedisUtil}
    import org.apache.kafka.clients.consumer.ConsumerRecord
    import org.apache.spark.streaming.dstream.{ DStream, InputDStream}
    import org.apache.spark.streaming.{ Seconds, StreamingContext}
    import org.json4s.jackson.Json
    import redis.clients.jedis.Jedis
    
    object  UserApp { 
      
      def main(args: Array[String]): Unit = { 
        
        val streamingContext: StreamingContext = new StreamingContext("local[*]","UserApp",Seconds(10))
        
        val ds: InputDStream[ConsumerRecord[String, String]] = MykafkaUtil.getKafkaStream(GmallConstants.KAFKA_TOPIC_USER_INFO, streamingContext)
      
        //无需转样例类，直接将DS中record的value值写入redis
      
        val ds1: DStream[String] = ds.map(record => record.value())
      
        //写入redis
        ds1.foreachRDD(rdd => { 
        
          rdd.foreachPartition(iter => { 
          
            //创建连接
            val jedisClient: Jedis = RedisUtil.getJedisClient
          
            iter.foreach(userInfoString => { 
            
              val userInfoKey = "user:" + JSON.parseObject(userInfoString).getString("id")
            
              jedisClient.set(userInfoKey , userInfoString)
            
            })
          
            jedisClient.close()
          
          })
        
        })
      
      
        streamingContext.start()
      
        streamingContext.awaitTermination()
      
      
      }
      
    }