Flink电商指标分析项目(4)_实时数据预处理

一时失言乱红尘 2022-11-06 01:57 159阅读 0赞

### 目录 ###

*   *  实时数据分析业务目标
     *  业务开发一般流程
     *  点击流日志实时数据预处理
     *   *  业务分析
         *  创建ClickLogWide样例类
         *  预处理：地址、时间字段拓宽
         *  预处理：isNew字段处理
         *  预处理：isHourNew字段处理
         *  预处理：isDayNew字段处理
         *  isMonthNew字段处理

## 实时数据分析业务目标 ##

*  完成点击流日志`数据预处理`业务开发
 *  完成实时频道`热点`分析业务开发
 *  完成实时频道`PV/UV`分析业务开发
 *  完成实时频道`用户新鲜度`分析业务开发
 *  完成实时频道`地域`分析业务开发

## 业务开发一般流程 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70]  
**一般流程**

1.  先对原始数据进行拓宽`预处理`
2.  将拓宽后的数据转换为要进行分析业务字段
3.  按照指定字段进行分组
4.  将组内数据划分到窗口中
5.  聚合计算
6.  将数据落地到hbase

`预处理`的过程非常关键，它是业务的源头，如果预处理做得很好，后续业务的开发会变得简单很多

## 点击流日志实时数据预处理 ##

### 业务分析 ###

为了方便后续分析，我们需要对点击流日志，使用Flink进行实时预处理。在原有点击流日志的基础上添加一些字段，方便进行后续业务功能的统计开发。  
以下为Kafka中消费得到的原始点击流日志字段：

<table> 
 <thead> 
  <tr> 
   <th>字段名</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>channelID</td> 
   <td>频道ID</td> 
  </tr> 
  <tr> 
   <td>categoryID</td> 
   <td>产品类别ID</td> 
  </tr> 
  <tr> 
   <td>produceID</td> 
   <td>产品ID</td> 
  </tr> 
  <tr> 
   <td>country</td> 
   <td>国家</td> 
  </tr> 
  <tr> 
   <td>province</td> 
   <td>省份</td> 
  </tr> 
  <tr> 
   <td>city</td> 
   <td>城市</td> 
  </tr> 
  <tr> 
   <td>network</td> 
   <td>网络方式</td> 
  </tr> 
  <tr> 
   <td>source</td> 
   <td>来源方式</td> 
  </tr> 
  <tr> 
   <td>browserType</td> 
   <td>浏览器类型</td> 
  </tr> 
  <tr> 
   <td>entryTime</td> 
   <td>进入网站时间</td> 
  </tr> 
  <tr> 
   <td>leaveTime</td> 
   <td>离开网站时间</td> 
  </tr> 
  <tr> 
   <td>userID</td> 
   <td>用户的ID</td> 
  </tr> 
 </tbody> 
</table>

我们需要在原有点击流日志字段基础上，再添加以下字段：

<table> 
 <thead> 
  <tr> 
   <th>字段名</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>count</td> 
   <td>用户访问的次数</td> 
  </tr> 
  <tr> 
   <td>timestamp</td> 
   <td>用户访问的时间</td> 
  </tr> 
  <tr> 
   <td>address</td> 
   <td>国家省份城市（拼接）</td> 
  </tr> 
  <tr> 
   <td>yearMonth</td> 
   <td>年月</td> 
  </tr> 
  <tr> 
   <td>yearMonthDay</td> 
   <td>年月日</td> 
  </tr> 
  <tr> 
   <td>yearMonthDayHour</td> 
   <td>年月日时</td> 
  </tr> 
  <tr> 
   <td>isNew</td> 
   <td>是否为访问某个频道的新用户</td> 
  </tr> 
  <tr> 
   <td>isHourNew</td> 
   <td>在某一小时内是否为某个频道的新用户</td> 
  </tr> 
  <tr> 
   <td>isDayNew</td> 
   <td>在某一天是否为某个频道的新用户</td> 
  </tr> 
  <tr> 
   <td>isMonthNew</td> 
   <td>在某一个月是否为某个频道的新用户</td> 
  </tr> 
 </tbody> 
</table>

我们不能直接从点击流日志中，直接计算得到上述后4个字段的值。而是需要在`hbase`中有一个`历史记录表`，来保存用户的历史访问状态才能计算得到。  
该历史记录表(`user_history`表)结构如下：

<table> 
 <thead> 
  <tr> 
   <th>列名</th> 
   <th>说明</th> 
   <th>示例</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>rowkey</td> 
   <td>用户ID:频道ID</td> 
   <td>10:220</td> 
  </tr> 
  <tr> 
   <td>userid</td> 
   <td>用户ID</td> 
   <td>10</td> 
  </tr> 
  <tr> 
   <td>channelid</td> 
   <td>频道ID</td> 
   <td>220</td> 
  </tr> 
  <tr> 
   <td>lastVisitedTime</td> 
   <td>最后访问时间（时间戳）</td> 
   <td>1553653555</td> 
  </tr> 
 </tbody> 
</table>

### 创建ClickLogWide样例类 ###

我们使用`ClickLogWide`样例类来保存拓宽后的点击流日志数据。直接`复制`原有的`ClickLog`样例类，然后给它加上下列额外的字段：  
**步骤**

1.  在`bean`包中创建`ClickLogWide`样例类
2.  复制原有`ClickLog样例类`
3.  添加以下字段
    
     *  用户访问的次数（count)
     *  用户访问的时间（timestamp)
     *  国家省份城市（拼接）（address)
     *  年月（yearMonth)
     *  年月日（yearMonthDay)
     *  年月日时（yearMonthDayHour)
     *  是否为访问某个频道的新用户（isNew)——0表示否，1表示是
     *  在某一小时内是否为某个频道的新用户（isHourNew)——0表示否
     *  在某一天是否为某个频道的新用户（isDayNew)——0表示否，1表示是，
     *  在某一个月是否为某个频道的新用户（isMonthNew)——0表示否，1表示是  
        **参考代码**  
        ClickLogWide.scala

// 频道ID(channelID)
    // 产品类别ID(categoryID)
    // 产品ID(produceID)
    // 国家(country)
    // 省份(province)
    // 城市(city)
    // 网络方式(network)
    // 来源方式(source)
    // 浏览器类型(browserType)
    // 进入网站时间(entryTime)
    // 离开网站时间(leaveTime)
    // 用户的ID(userID)
    // 用户访问的次数（count)
    // 用户访问的时间（timestamp)
    // 国家省份城市（拼接）（address)
    // 年月（yearMonth)
    // 年月日（yearMonthDay)
    // 年月日时（yearMonthDayHour)
    // 是否为访问某个频道的新用户（isNew)——0表示否，1表示是
    // 在某一小时内是否为某个频道的新用户（isHourNew)——0表示否
    // 在某一天是否为某个频道的新用户（isDayNew)——0表示否，1表示是，
    // 在某一个月是否为某个频道的新用户（isMonthNew)——0表示否，1表示是
    case class ClickLog(
                         var channelID: String,
                         var categoryID: String,
                         var produceID: String,
                         var country: String,
                         var province: String,
                         var city: String,
                         var network: String,
                         var source: String,
                         var browserType: String,
                         var entryTime: String,
                         var leaveTime: String,
                         var userID: String,
                         var count: Long,
                         var timestamp: Long,
                         var address: String,
                         var yearMonth: String,
                         var yearMonthDay: String,
                         var yearMonthDayHour: String,
                         var isNew: Int,
                         var isHourNew: Int,
                         var isDayNew: Int,
                         var isMonthNew: Int
                       )

### 预处理：地址、时间字段拓宽 ###

为了方便后续进行数据分析，我们需要对地址、时间字段进行拓宽。我们先来拓宽以下几个字段：

<table> 
 <thead> 
  <tr> 
   <th>字段名</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>address</td> 
   <td>国家、省份、城市（拼接）</td> 
  </tr> 
  <tr> 
   <td>yearMonth</td> 
   <td>年月</td> 
  </tr> 
  <tr> 
   <td>yearMonthDay</td> 
   <td>年月日</td> 
  </tr> 
  <tr> 
   <td>yearMonthDayHour</td> 
   <td>年月日时</td> 
  </tr> 
 </tbody> 
</table>

**步骤**

1.  对水印数据流进行预处理
2.  拼接拓宽字段
3.  使用`SimpleDateFormat+FastDateFormat`来获取格式化的时间（FastDateFormat是`线程安全`的）
4.  构建`ClickLogWide`样例类
5.  打印预处理数据
6.  测试

**实现**

*  在`task`包下创建`PreprocessTask`
 *  添加`process`接收水印流数据
 *  使用`map`算子进行数据转换
 *  构建以下拓宽字段
    
     *  国家省份城市（拼接）（address)
     *  年月（yearMonth)
     *  年月日（yearMonthDay)
     *  年月日时（yearMonthDayHour)
 *  编写工具方法`timestamp2Str`，传入时间戳和日期格式，使用`FastDateFormat`将时间戳转换为指定格式的日期
 *  分别使用`timestamp2Str`获取`年月`、`年月日`、`年月日时`字段
 *  构建`ClickLogWide`样例类
 *  在`App`中调用预处理任务的`process`方法，并打印测试  
    **示例代码**

import com.itheima.realprocess.bean.{ ClickLogWide, Message}
    import org.apache.commons.lang.time.FastDateFormat
    import org.apache.flink.streaming.api.scala.DataStream
    import org.apache.flink.api.scala._
    
    /** * 预处理任务 */
    object PreprocessTask { 
    
      def process(watermarkDataStream: DataStream[Message]) = { 
    
        val clickWideDataStream = watermarkDataStream.map { 
          msg => { 
            //转换时间
            val yearMonth = FastDateFormat.getInstance("yyyyMM").format(msg.timestamp)
            val yearMonthDay = FastDateFormat.getInstance("yyyyMMdd").format(msg.timestamp)
            val yearMonthDayHour = FastDateFormat.getInstance("yyyyMMddHH").format(msg.timestamp)
    
            //转换地区
            val address = msg.clickLog.country + msg.clickLog.province + msg.clickLog.city
    
            ClickLogWide(
              msg.clickLog.channelID,
              msg.clickLog.categoryID,
              msg.clickLog.produceID,
              msg.clickLog.country,
              msg.clickLog.province,
              msg.clickLog.city,
              msg.clickLog.network,
              msg.clickLog.source,
              msg.clickLog.browserType,
              msg.clickLog.entryTime,
              msg.clickLog.leaveTime,
              msg.clickLog.userID,
              msg.count,
              msg.timestamp,
              address,
              yearMonth,
              yearMonthDay,
              yearMonthDayHour,
              0,
              0,
              0,
              0
            )
    
          }
        }
        clickWideDataStream
      }
    }

`App.scala`中添加执行任务代码

//数据的预处理
        val clickLogWideDataStream = PreprocessTask.process(watermarkDataStream)
        clickLogWideDataStream.print()

### 预处理：isNew字段处理 ###

isNew字段是判断某个`用户ID`，是否已经访问过`某个频道`。  
**实现思路**

1.  先把要拓宽的字段`isNew`、`isHourNew`、`isDayNew`、`isMonthNew`都创建出来，初始化为0
2.  从hbase查询rowkey为`userid:channlid`查询`user_history`中`userid`列的数据
3.  判断`userid`列数据是否为空

*  如果为空
    
     *  设置`isNew`字段为`1`，表示是新用户
     *  将其他的`isHourNew`、`isDayNew`、`isMonthNew`字段都设置为1
     *  将该用户数据添加到`user_history`表中
 *  如果不为空
    
     *  设置`isNew`字段为0，表示是老用户
     *  更新`user_history`表中的`lastVisitedTime`列  
        **user\_history表的列**
 *  用户ID:频道ID(rowkey)
 *  用户ID(userid)
 *  频道ID(channelid)
 *  最后访问时间（时间戳）(lastVisitedTime)

**步骤**

1.  创建一个object `RealPreprogressTask`，用来编写预处理的scala代码  
    2 创建一个方法`process`，接收水印流数据  
    3 使用map算子按照实现思路进行处理，计算得到`isNew`字段，然后返回  
    4 打印测试，查看hbase中是否已经成功创建`user_history`表并保存了用户数据

### 预处理：isHourNew字段处理 ###

`isHourNew`字段表示，用户虽然不是第一次访问某个频道，但是用户在`某一个小时`内，是第一次访问该频道。  
我们通过时间比较来判断：

1.  如果当前用户的访问时间是：2020-03-27 11点
2.  历史记录表中，最后一次的访问时间是：2020-03-27 10点，isHourNew为`1`（11点确实没访问）
3.  历史记录表中，最后一次的访问时间是：2020-03-27 11点，isHourNew为`0`  
    所以，我们根据时间来进行判断。  
    **步骤**
4.  从`user_history`表中获取`lastVisitedTime`字段
5.  将`lastVisitedTime`字段格式化为`年月日时`格式
6.  和当前点击流日志的`年/月/日/时`字段比较
    
     *  如果当前时间字段 > `lastVisited`，则`isHourNew`为1
     *  否则`isHourNew`为0

### 预处理：isDayNew字段处理 ###

与`isHourNew`字段处理类似，只不过需要使用`年月日`来进行比较

### isMonthNew字段处理 ###

与`isHourNew`字段处理类似，只不过需要使用`年月`来进行比较  
**参考代码**

import com.itheima.realprocess.bean.{ ClickLogWide, Message}
    import com.itheima.realprocess.util.HBaseUtil
    import org.apache.commons.lang.StringUtils
    import org.apache.commons.lang.time.FastDateFormat
    import org.apache.flink.streaming.api.scala.DataStream
    import org.apache.flink.api.scala._
    
    
    /** * 预处理任务 */
    object PreprocessTask { 
    
      def process(watermarkDataStream: DataStream[Message]) = { 
    
        val clickWideDataStream = watermarkDataStream.map { 
          msg => { 
            //转换时间
            val yearMonth = FastDateFormat.getInstance("yyyyMM").format(msg.timestamp)
            val yearMonthDay = FastDateFormat.getInstance("yyyyMMdd").format(msg.timestamp)
            val yearMonthDayHour = FastDateFormat.getInstance("yyyyMMddHH").format(msg.timestamp)
    
            //转换地区
            val address = msg.clickLog.country + msg.clickLog.province + msg.clickLog.city
    
            val isNewTuple = isNewProcess(msg)
    
    
            ClickLogWide(
              msg.clickLog.channelID,
              msg.clickLog.categoryID,
              msg.clickLog.produceID,
              msg.clickLog.country,
              msg.clickLog.province,
              msg.clickLog.city,
              msg.clickLog.network,
              msg.clickLog.source,
              msg.clickLog.browserType,
              msg.clickLog.entryTime,
              msg.clickLog.leaveTime,
              msg.clickLog.userID,
              msg.count,
              msg.timestamp,
              address,
              yearMonth,
              yearMonthDay,
              yearMonthDayHour,
              isNewTuple._1,
              isNewTuple._2,
              isNewTuple._3,
              isNewTuple._4
            )
    
          }
        }
        clickWideDataStream
      }
    
      private def isNewProcess(msg:Message)={ 
    
        //1.定义4个变量，初始化为0
        var isNew = 0
        var isHourNew = 0
        var isDayNew = 0
        var isMonthNew = 0
    
        //2.从hbase中查询用户记录，如果有记录，再去判断其他事件，如果没有记录，则证明是新用户
        // 表名, 渠道用户历史表
        val tableName = "channel_user_history"
        // 列族名
        val cfName = "info"
        // 拼接rowkey ------> 用户ID:频道ID(rowkey)
        val rowkey: String = msg.clickLog.userID + ":" + msg.clickLog.channelID
        // 用户ID(userid)
        val userIdColName = "userid"
        // 频道ID(channelid)
        val channelIdColName = "channelId"
        // 最后访问时间（时间戳）(lastVisitedTime)
        val lastVisitedTimeColName = "lastVisitedTime"
    
        val userId = HBaseUtil.getData(tableName, rowkey, cfName, userIdColName)
        val lastVisitedTime = HBaseUtil.getData(tableName, rowkey, cfName, lastVisitedTimeColName)
    
        //如果userid为空，则该用户一定是新用户
        if(StringUtils.isBlank(userId)){ 
          isNew = 1
          isHourNew = 1
          isDayNew = 1
          isMonthNew = 1
    
          //保存用户的访问记录到`user_history`
          HBaseUtil.putMapData(tableName,rowkey,cfName,Map(
            userIdColName->msg.clickLog.userID,
            channelIdColName->msg.clickLog.channelID,
            lastVisitedTimeColName->msg.timestamp
          ))
    
    
        }else{ 
          isNew = 0
          //其他字段需要进行时间戳的比对 201912 > 2011912
          isHourNew = compareDate(msg.timestamp,lastVisitedTime.toLong,"yyyyMMddHH")
          isDayNew = compareDate(msg.timestamp,lastVisitedTime.toLong,"yyyyMMdd")
          isMonthNew = compareDate(msg.timestamp,lastVisitedTime.toLong,"yyyyMM")
          //更新‘user_history’的用户的时间戳
          HBaseUtil.putData(tableName,rowkey,cfName,lastVisitedTimeColName,msg.timestamp.toString)
        }
    
        (isNew,isHourNew,isDayNew,isMonthNew)
      }
    
      //201912 > 2011912
      /** * 比对时间 * @param currentTime 当前时间 * @param historyTime 历史时间 * @param format 时间格式 yyyyMM yyyyMMdd */
      def compareDate(currentTime:Long,historyTime:Long,format:String): Int ={ 
    
        val currentTimeStr = timestamp2str(currentTime, format)
        val historyTimeStr = timestamp2str(historyTime, format)
    
        //比对字符串的大小，如果当前时间>历史时间 返回1
    
        var result = currentTimeStr.compareTo(historyTimeStr)
    
        if (result>0){ 
          result = 1
        }else{ 
          result = 0
        }
        result
      }
    
      def timestamp2str(timestamp:Long,format:String):String={ 
        FastDateFormat.getInstance(format).format(timestamp)
      }
    
      def main(args: Array[String]): Unit = { 
        println(compareDate(1578104383523L, 1568304383523L, "yyyyMMdd"))
      }
    }

注意：
    
    日期格式
    
    1. yyyy :	年
    2. MM :     月
    3. dd:       日
    4. HH:      24小时

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70]: /images/20221023/9c7e64074e4f4be88308955040435e47.png