Flink电商指标分析项目(5)_实时频道热点分析业务开发

╰+攻爆jí腚メ 2022-11-06 03:00 215阅读 0赞

### 目录 ###

*  实时频道热点分析业务开发
 *   *  业务介绍
     *  业务开发
 *  实时频道PV/UV分析
 *   *  业务介绍
     *  小时维度PV/UV业务开发
     *  天维度PV/UV业务开发
     *  小时/天/月维度PV/UV业务开发
 *  实时频道用户新鲜度分析
 *   *  业务介绍
     *  业务开发
     *  模板方法提取公共类
 *  实时频道地域分析业务开发
 *   *  业务介绍
     *  业务开发
 *  实时运营商分析业务开发
 *   *  业务介绍
     *  业务开发
 *  实时频道浏览器分析业务开发
 *   *  业务介绍
     *  业务开发
 *  数据输不出来请看这

# 实时频道热点分析业务开发 #

## 业务介绍 ##

频道热点，就是要统计频道被访问（点击）的数量。  
分析得到以下的数据：

需要将历史的点击数据进行累加

## 业务开发 ##

**步骤**

1.  创建实时热点样例类，专门用来计算实时热点的数据
2.  将预处理后的数据，`转换`为要分析出来的数据（频道、访问次数）样例类
3.  按照`频道`进行分组（分流）
4.  划分时间窗口（3秒一个窗口）
5.  进行合并计数统计
6.  打印测试
7.  将计算后的数据下沉到Hbase  
    **实现**
8.  创建一个`ChannelRealHotTask`单例对象
9.  添加一个`ChannelRealHot`样例类，它封装要统计的两个业务字段：频道ID（channelID）、访问数量（visited）
10. 在`ChannelRealHotTask`中编写一个`process`方法，接收预处理后的`DataStream`
11. 使用map算子，将`ClickLog`对象转换为`ChannelRealHot`
12. 按照频道ID进行分流
13. 划分时间窗口（3秒一个窗口）
14. 执行reduce合并计算
15. 将合并后的数据下沉到hbase
    
     *  判断hbase中是否已经存在结果记录
     *  若存在，则获取后进行累加
     *  若不存在，则直接写入  
        **参考代码**

import com.itheima.realprocess.bean.ClickLogWide
    import com.itheima.realprocess.util.HBaseUtil
    import org.apache.commons.lang.{ StringEscapeUtils, StringUtils}
    import org.apache.flink.streaming.api.scala.{ DataStream, KeyedStream, WindowedStream}
    import org.apache.flink.api.scala._
    import org.apache.flink.streaming.api.functions.sink.SinkFunction
    import org.apache.flink.streaming.api.windowing.time.Time
    import org.apache.flink.streaming.api.windowing.windows.TimeWindow
    
    case class ChannelRealHot(
                             var channelId:String,
                             var visited:Long
                             )
    
    /** * 频道热点分析 * * 1.字段转换 * 2.分组 * 3.时间窗口 * 4.聚合 * 5.落地hbase */
    object ChannelRealHotTask { 
    
      def process(clickLogWideDataStream: DataStream[ClickLogWide])={ 
    
        //1.字段转换 channelid visited
        val realHotDataStream = clickLogWideDataStream.map { 
          clickLogWide: ClickLogWide =>
            ChannelRealHot(clickLogWide.channelID, clickLogWide.count)
        }
    
        //2.分组
        val keyedStream: KeyedStream[ChannelRealHot, String] = realHotDataStream.keyBy(_.channelId)
    
        //3.时间窗口
        val windowedStream: WindowedStream[ChannelRealHot, String, TimeWindow] = keyedStream.timeWindow(Time.seconds(3))
    
        //4.聚合
        val reduceDataStream = windowedStream.reduce { 
          (t1: ChannelRealHot, t2: ChannelRealHot) => { 
            ChannelRealHot(t1.channelId, t1.visited + t2.visited)
          }
        }
       // reduceDataStream
        //落地hbase
        reduceDataStream.addSink(new SinkFunction[ChannelRealHot] { 
          override def invoke(value: ChannelRealHot): Unit = { 
    
            //hbase相关字段
            val tableName = "channel"
            val clfName = "info"
            val channelIdColumn = "channelId"
            val visitedColumn = "visited"
            val rowkey = value.channelId
    
            //查询hbase，获取相关记录
            val visitedValue = HBaseUtil.getData(tableName, rowkey, clfName, visitedColumn)
            //创建总数的临时变量
            var totolCount:Long = 0
            if(StringUtils.isBlank(visitedValue)){ 
              totolCount = value.visited
            }else{ 
              totolCount = visitedValue.toLong + value.visited
            }
    
            //保存数据
            HBaseUtil.putMapData(tableName,rowkey,clfName,Map(
              channelIdColumn -> value.channelId,
              visitedColumn -> totolCount.toString
            ))
          }
        })
      }
    
    }

# 实时频道PV/UV分析 #

针对频道的PV、UV进行不同时间维度的分析。有以下三个维度：

*  小时
 *  天
 *  月  
    我们会分别来实现不同维度的分析。

## 业务介绍 ##

**PV(访问量)**  
即Page View，页面刷新一次算一次。  
**UV(独立访客)**  
即Unique Visitor，指定时间内相同的客户端只被计算一次  
统计分析后得到的数据如下所示：

## 小时维度PV/UV业务开发 ##

**步骤**

1.  创建频道PV、UV样例类
2.  将预处理后的数据，`转换`为要分析出来的数据（频道、PV、UV）样例类
3.  按照`频道`和`时间`进行分组（分流）
4.  划分时间窗口（3秒一个窗口）
5.  进行合并计数统计
6.  打印测试
7.  将计算后的数据下沉到Hbase  
    **实现**
8.  创建一个`ChannelPvUvTask`单例对象
9.  添加一个`ChannelPvUv`样例类，它封装要统计的四个业务字段：频道ID（channelID）、年月日时、PV、UV
10. 在`ChannelPvUvTask`中编写一个`processHourDim`方法，接收预处理后的`DataStream`
11. 使用map算子，将`ClickLog`对象转换为`ChannelPvUv`
12. 按照`频道ID`、`年月日时`进行分流
13. 划分时间窗口（3秒一个窗口）
14. 执行reduce合并计算
15. 打印测试
16. 将合并后的数据下沉到hbase
    
     *  判断hbase中是否已经存在结果记录
     *  若存在，则获取后进行累加
     *  若不存在，则直接写入  
        **参考代码**

// 抽取一个公共的trait, 所有的任务都来实现它
    trait BaseTask { 
      def process(clickLogWideDataStream: DataStream[ClickLogWide]):Any
    }
    
    
    //添加一个`ChannelPvUv`样例类，它封装要统计的四个业务字段：频道ID（channelID）、年月日时、PV、UV
    case class ChannelPvUv(
                            val channelID: String,
                            var yearMonthDayHour: String,
                            var pv: Long,
                            var uv: Long
                          )
    
    object ChannelPvUvTask extends BaseTask { 
    
      override def process(clickLogWideDataStream: DataStream[ClickLogWide]) = { 
    
        // 1. 使用map算子，将`ClickLog`对象转换为`ChannelPvUv`
        val channelPvUvDataStream: DataStream[ChannelPvUv] = clickLogWideDataStream.map { 
          clickLogWide =>
            ChannelPvUv(clickLogWide.channelID, clickLogWide.yearMonthDayHour, clickLogWide.count, clickLogWide.isHourNew)
        }
    
        // 2. 根据channelID+年月日时 分组 keyBy
        val keyedStream: KeyedStream[ChannelPvUv, String] = channelPvUvDataStream.keyBy { 
          channelPvUv => channelPvUv.channelID + channelPvUv.yearMonthDayHour
        }
    
        // 3. 设置时间窗口
        val windowDataStream: WindowedStream[ChannelPvUv, String, TimeWindow] = keyedStream.timeWindow(Time.seconds(3))
    
        // 4. 聚合
        val reduceDataStream: DataStream[ChannelPvUv] = windowDataStream.reduce { 
          (t1, t2) =>
            ChannelPvUv(t1.channelID, t1.yearMonthDayHour, t1.pv + t2.pv, t1.uv + t2.uv)
        }
    
        // 5. 落地HBase
        reduceDataStream.addSink(new SinkFunction[ChannelPvUv] { 
    
          override def invoke(value: ChannelPvUv): Unit = { 
            val tableName = "channel_pvuv"
            val cfName = "info"
            val channelIdColName = "channelID"
            val yearMonthDayHourColName = "yearMonthDayHour"
            val pvColName = "pv"
            val uvColName = "uv"
            val rowkey = value.channelID + ":" + value.yearMonthDayHour
    
            var totalPv = 0L
            var totalUv = 0L
    
            // - 判断hbase中是否已经存在结果记录
            val pvValue: String = HBaseUtil.getData(tableName, rowkey, cfName, pvColName)
            val uvValue: String = HBaseUtil.getData(tableName, rowkey, cfName, uvColName)
            // - 若存在，则获取后进行累加
            if(!StringUtils.isBlank(pvValue)) { 
              totalPv = pvValue.toLong + value.pv
            }
            // - 若不存在，则直接写入
            else { 
              totalPv = value.pv
            }
    
            // - 若存在，则获取后进行累加
            if(!StringUtils.isBlank(uvValue)) { 
              totalUv = uvValue.toLong + value.uv
            }
            // - 若不存在，则直接写入
            else { 
              totalUv = value.uv
            }
    
            HBaseUtil.putMapData(tableName,rowkey,cfName,Map(
              channelIdColName -> value.channelID,
              yearMonthDayHourColName -> value.yearMonthDayHour,
              pvColName -> totalPv.toString,
              uvColName -> totalUv.toString
            ))
    
          }
        })
      }
    }

## 天维度PV/UV业务开发 ##

按天的维度来统计PV、UV与按小时维度类似，就是分组字段不一样。可以直接复制按小时维度的PV/UV，然后修改即可。

## 小时/天/月维度PV/UV业务开发 ##

但是，其实上述代码，都是一样的。我们可以将`小时`、`天`、`月`三个时间维度的数据放在一起来进行分组  
**思路**

1.  每一条ClickLog生成三个维度的`ChannelPvUv`，分别用于三个维度的统计

*  ChannelPvUv --> 小时维度
 *  ChannelPvUv --> 天维度
 *  ChannelPvUv --> 月维度

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70]  
**实现**

1.  使用`flatmap`算子，将`ClickLog`转换为三个`ChannelPvUv`
2.  重新运行测试  
    **参考代码**

def process(clicklogWideDataStream:DataStream[ClickLogWide]) = { 
        ...
        val channelPvUvDataStream: DataStream[ChannelPvUv] = clicklogWideDataStream.flatMap { 
          clicklog =>
            List(
              ChannelPvUv(clicklog.channelID, clicklog.yearMonthDayHour, clicklog.count, clicklog.isHourNew),
              ChannelPvUv(clicklog.channelID, clicklog.yearMonthDay, clicklog.count, clicklog.isDayNew),
              ChannelPvUv(clicklog.channelID, clicklog.yearMonth, clicklog.count, clicklog.isMonthNew)
            )
        }
        ...
      }

# 实时频道用户新鲜度分析 #

## 业务介绍 ##

用户新鲜度即分析网站每小时、每天、每月活跃的新老用户占比  
可以通过新鲜度：

*  从宏观层面上了解每天的新老用户比例以及来源结构
 *  当天新增用户与当天`推广行为`是否相关  
    统计分析要得到的数据如下：

## 业务开发 ##

**步骤**

1.  创建频道新鲜度样例类，包含以下字段（频道、时间、新用户、老用户）
2.  将预处理后的数据，转换为新鲜度样例类
3.  按照`频道`和`时间`进行分组（分流）
4.  划分时间窗口（3秒一个窗口）
5.  进行合并计数统计
6.  打印测试
7.  将计算后的数据下沉到Hbase  
    **实现**
8.  创建一个`ChannelFreshnessTask`单例对象
9.  添加一个`ChannelFreshness`样例类，它封装要统计的四个业务字段：频道ID（channelID）、日期（date）、新用户（newCount）、老用户（oldCount）
10. 在`ChannelFreshnessTask`中编写一个`process`方法，接收预处理后的`DataStream`
11. 使用`flatMap`算子，将`ClickLog`对象转换为三个不同时间维度`ChannelFreshness`
12. 按照`频道ID`、`日期`进行分流
13. 划分时间窗口（3秒一个窗口）
14. 执行reduce合并计算
15. 打印测试
16. 将合并后的数据下沉到hbase
    
     *  准备hbase的表名、列族名、rowkey名、列名
    
     *  判断hbase中是否已经存在结果记录
     *  若存在，则获取后进行累加
     *  若不存在，则直接写入

注意：
    这个地方，老用户需要注意处理，因为如果不进行判断，就会计算重复的一些用户访问数据
    1. 新用户就是根据clicklog拓宽后的isNew来判断
    2. 老用户需要判断
      * 如果isNew是0，且isHourNew为1/isDayNew为1、isMonthNew为1，则进行老用户为1
      * 否则为0

参考代码

import com.itheima.realprocess.bean.ClickLogWide
    import com.itheima.realprocess.util.HBaseUtil
    import org.apache.commons.lang.StringUtils
    import org.apache.flink.streaming.api.scala.{ DataStream, KeyedStream, WindowedStream}
    import org.apache.flink.api.scala._
    import org.apache.flink.streaming.api.functions.sink.SinkFunction
    import org.apache.flink.streaming.api.windowing.time.Time
    import org.apache.flink.streaming.api.windowing.windows.TimeWindow
    
    case class ChannelFreshness(
            var channelID:String,
            var date:String,
            var newCount:Long,
            var oldCount:Long
            )
    
            object ChannelFreshnessTask { 
    
              def process(clickLogWideDataStream: DataStream[ClickLogWide])={ 
                //转换
                val mapDataStream: DataStream[ChannelFreshness] = clickLogWideDataStream.flatMap { 
                  clickLog => { 
                    //如果是老用户 只有在他第一次来的时候 计数为1
                    val isOld = (isNew: Int, isDateNew: Int) => if (isNew == 0 && isDateNew == 1) 1 else 0
    
                    List(
                      ChannelFreshness(clickLog.channelID, clickLog.yearMonthDayHour, clickLog.isNew, isOld(clickLog.isNew, clickLog
                        .yearMonthDayHour.toInt)),
                      ChannelFreshness(clickLog.channelID, clickLog.yearMonthDayHour, clickLog.isNew, isOld(clickLog.isNew, clickLog
                        .yearMonthDay.toInt)),
                      ChannelFreshness(clickLog.channelID, clickLog.yearMonthDayHour, clickLog.isNew, isOld(clickLog.isNew, clickLog
                        .yearMonth.toInt))
                    )
                  }
                }
    
                val keyedStream: KeyedStream[ChannelFreshness, String] = mapDataStream.keyBy { 
                  freshness => (freshness.channelID + freshness.date)
                }
    
                //时间窗口
                val windowStream: WindowedStream[ChannelFreshness, String, TimeWindow] = keyedStream.timeWindow(Time.seconds(3))
    
                //聚合
                val reduceDataStream: DataStream[ChannelFreshness] = windowStream.reduce { 
                  (t1, t2) => { 
                    ChannelFreshness(t1.channelID, t1.date, t1.newCount + t2.newCount, t1.oldCount + t2.oldCount)
                  }
                }
    
    
                //落地
                reduceDataStream.addSink(new SinkFunction[ChannelFreshness] { 
                  override def invoke(value: ChannelFreshness): Unit = { 
                    //创建Hbase相关变量
                    val tableName = "channel_freshness"
                    val clfName = "info"
                    val rowkey = value.channelID+":"+value.date
    
                    val channelIDColumn = "channelId"
                    val dateColumn = "date"
                    val newCountColumn = "newCount"
                    val oldCountColumn = "oldCount"
    
                    //查询历史数据
                    val resultMap = HBaseUtil.getMapData(tableName, rowkey, clfName, List(newCountColumn, oldCountColumn))
    
                    //累加
                    var totalNewCount = 0L
                    var totalOldCount = 0L
    
    
                    if (resultMap != null && StringUtils.isNotBlank(resultMap.getOrElse(newCountColumn,""))){ 
                      totalNewCount = resultMap(newCountColumn).toLong+value.newCount
                    }else{ 
                      totalNewCount = value.newCount
                    }
    
                    if (resultMap != null && StringUtils.isNotBlank(resultMap.getOrElse(oldCountColumn,""))){ 
                      totalOldCount = resultMap(oldCountColumn).toLong+value.oldCount
                    }else{ 
                      totalNewCount = value.newCount
                    }
    
                    //保存数据
                    HBaseUtil.putMapData(tableName,rowkey,clfName,Map(
                      channelIDColumn -> value.channelID,
                      dateColumn -> value.date,
                      newCountColumn -> totalNewCount.toString,
                      oldCountColumn -> totalOldCount.toString
            ))
          }
        })
    
      }
    
    
    }

## 模板方法提取公共类 ##

模板方法模式是在父类中定义算法的骨架，把具体实延迟到子类中去，可以在不改变一个算法的结构时可重定义该算法的某些步骤。  
**BaseTask.scala**

import com.itheima.realprocess.bean.ClickLogWide
    import org.apache.flink.streaming.api.scala.{ DataStream, KeyedStream, WindowedStream}
    import org.apache.flink.streaming.api.windowing.windows.TimeWindow
    
    trait BaseTask[T] { 
    
      //转换
      def map(clickLogWideDataStream: DataStream[ClickLogWide]):DataStream[T]
    
      //分组
      def keyBy(mapDataStream:DataStream[T]):KeyedStream[T, String]
    
      //时间窗口
      def timeWindow(keyedStream: KeyedStream[T,String]):WindowedStream[T, String, TimeWindow]
    
      //聚合
      def reduce(windowedStream: WindowedStream[T, String, TimeWindow]):DataStream[T]
    
      //落地HBase
      def sink2HBase(reduceDataStream:DataStream[T])
    
      def process(clickLogWideDataStream: DataStream[ClickLogWide]):Unit={ 
        val mapDataStream = map(clickLogWideDataStream)
        val keyedStream = keyBy(mapDataStream)
        val windowedStream = timeWindow(keyedStream)
        val reduceStream = reduce(windowedStream)
        sink2HBase(reduceStream)
      }
    }

**改造后的代码:**

import com.itheima.realprocess.bean.ClickLogWide
    import com.itheima.realprocess.util.HBaseUtil
    import org.apache.commons.lang.StringUtils
    import org.apache.flink.streaming.api.scala.{ DataStream, KeyedStream, WindowedStream}
    import org.apache.flink.api.scala._
    import org.apache.flink.streaming.api.functions.sink.SinkFunction
    import org.apache.flink.streaming.api.windowing.time.Time
    import org.apache.flink.streaming.api.windowing.windows.TimeWindow
    
    case class ChannelFreshness(
            var channelID:String,
            var date:String,
            var newCount:Long,
            var oldCount:Long
            )
    
    object ChannelFreshnessTask extends BaseTask[ChannelFreshness]{ 
      override def map(clickLogWideDataStream: DataStream[ClickLogWide]): DataStream[ChannelFreshness] = { 
        //转换
        val mapDataStream: DataStream[ChannelFreshness] = clickLogWideDataStream.flatMap { 
          clickLog => { 
            //如果是老用户 只有在他第一次来的时候 计数为1
            val isOld = (isNew: Int, isDateNew: Int) => if (isNew == 0 && isDateNew == 1) 1 else 0
    
            List(
              ChannelFreshness(clickLog.channelID, clickLog.yearMonthDayHour, clickLog.isNew, isOld(clickLog.isNew, clickLog
                .yearMonthDayHour.toInt)),
              ChannelFreshness(clickLog.channelID, clickLog.yearMonthDayHour, clickLog.isNew, isOld(clickLog.isNew, clickLog
                .yearMonthDay.toInt)),
              ChannelFreshness(clickLog.channelID, clickLog.yearMonthDayHour, clickLog.isNew, isOld(clickLog.isNew, clickLog
                .yearMonth.toInt))
            )
          }
        }
        mapDataStream
      }
    
      override def keyBy(mapDataStream: DataStream[ChannelFreshness]): KeyedStream[ChannelFreshness, String] = { 
        mapDataStream.keyBy { 
          freshness =>
            freshness.channelID + freshness.date
        }
      }
    
      override def timeWindow(keyedStream: KeyedStream[ChannelFreshness, String]): WindowedStream[ChannelFreshness,
        String, TimeWindow] = { 
        keyedStream.timeWindow(Time.seconds(3))
      }
    
      override def reduce(windowedStream: WindowedStream[ChannelFreshness, String, TimeWindow]): DataStream[ChannelFreshness] = { 
        windowedStream.reduce { 
          (t1, t2) => { 
            ChannelFreshness(t1.channelID, t1.date, t1.newCount + t2.newCount, t1.oldCount + t2.oldCount)
          }
        }
      }
    
      override def sink2HBase(reduceDataStream: DataStream[ChannelFreshness]): Unit = { 
        //落地
        reduceDataStream.addSink(new SinkFunction[ChannelFreshness] { 
          override def invoke(value: ChannelFreshness): Unit = { 
            //创建Hbase相关变量
            val tableName = "channel_freshness"
            val clfName = "info"
            val rowkey = value.channelID+":"+value.date
    
            val channelIDColumn = "channelId"
            val dateColumn = "date"
            val newCountColumn = "newCount"
            val oldCountColumn = "oldCount"
    
            //查询历史数据
            val resultMap = HBaseUtil.getMapData(tableName, rowkey, clfName, List(newCountColumn, oldCountColumn))
    
            //累加
            var totalNewCount = 0L
            var totalOldCount = 0L
    
    
            if (resultMap != null && StringUtils.isNotBlank(resultMap.getOrElse(newCountColumn,""))){ 
              totalNewCount = resultMap(newCountColumn).toLong+value.newCount
            }else{ 
              totalNewCount = value.newCount
            }
    
            if (resultMap != null && StringUtils.isNotBlank(resultMap.getOrElse(oldCountColumn,""))){ 
              totalOldCount = resultMap(oldCountColumn).toLong+value.oldCount
            }else{ 
              totalNewCount = value.newCount
            }
    
            //保存数据
            HBaseUtil.putMapData(tableName,rowkey,clfName,Map(
              channelIDColumn -> value.channelID,
              dateColumn -> value.date,
              newCountColumn -> totalNewCount.toString,
              oldCountColumn -> totalOldCount.toString
            ))
          }
        })
    
    
      }
    }

# 实时频道地域分析业务开发 #

## 业务介绍 ##

通过地域分析，可以帮助查看地域相关的PV/UV、用户新鲜度。  
**需要分析出来指标**

*  PV
 *  UV
 *  新用户
 *  老用户  
    **需要分析的维度**
 *  地域（国家省市）——这里为了节省时间，只分析市级的地域维度，其他维度大家可以自己来实现
 *  时间维度（时、天、月）  
    统计分析后的结果如下：

<table> 
 <thead> 
  <tr> 
   <th>频道ID</th> 
   <th>地域（国/省/市）</th> 
   <th>时间</th> 
   <th>PV</th> 
   <th>UV</th> 
   <th>新用户</th> 
   <th>老用户</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>频道1</td> 
   <td>中国北京市朝阳区</td> 
   <td>201809</td> 
   <td>1000</td> 
   <td>300</td> 
   <td>123</td> 
   <td>171</td> 
  </tr> 
  <tr> 
   <td>频道1</td> 
   <td>中国北京市朝阳区</td> 
   <td>20180910</td> 
   <td>512</td> 
   <td>123</td> 
   <td>23</td> 
   <td>100</td> 
  </tr> 
  <tr> 
   <td>频道1</td> 
   <td>中国北京市朝阳区</td> 
   <td>2018091010</td> 
   <td>100</td> 
   <td>41</td> 
   <td>11</td> 
   <td>30</td> 
  </tr> 
 </tbody> 
</table>

## 业务开发 ##

**步骤**

1.  创建频道地域分析样例类（频道、地域（国省市）、时间、PV、UV、新用户、老用户）
2.  将预处理后的数据，使用`flatMap`转换为样例类
3.  按照`频道`、`时间`、`地域`进行分组（分流）
4.  划分时间窗口（3秒一个窗口）
5.  进行合并计数统计
6.  打印测试
7.  将计算后的数据下沉到Hbase

**实现**

1.  创建一个`ChannelAreaTask`单例对象
2.  添加一个`ChannelArea`样例类，它封装要统计的四个业务字段：频道ID（channelID）、地域（area）、日期（date）pv、uv、新用户（newCount）、老用户（oldCount）
3.  在`ChannelAreaTask`中编写一个`process`方法，接收预处理后的`DataStream`
4.  使用`flatMap`算子，将`ClickLog`对象转换为三个不同时间维度`ChannelArea`
5.  按照`频道ID`、`时间`、`地域`进行分流
6.  划分时间窗口（3秒一个窗口）
7.  执行reduce合并计算
8.  打印测试
9.  将合并后的数据下沉到hbase
    
     *  准备hbase的表名、列族名、rowkey名、列名
     *  判断hbase中是否已经存在结果记录
     *  若存在，则获取后进行累加
     *  若不存在，则直接写入  
        **参考代码**  
        ChannelAreaTask.scala

import com.itheima.realprocess.bean.ClickLogWide
    import com.itheima.realprocess.util.HBaseUtil
    import org.apache.commons.lang.StringUtils
    import org.apache.flink.streaming.api.scala.{ DataStream, KeyedStream, WindowedStream}
    import org.apache.flink.streaming.api.windowing.windows.TimeWindow
    import org.apache.flink.api.scala._
    import org.apache.flink.streaming.api.windowing.time.Time
    
    case class ChannelArea(
                          var channelId:String,
                          var area:String,
                          var date:String,
                          var pv:Long,
                          var uv:Long,
                          var newCount:Long,
                          var oldCount:Long
                          )
    
    object ChannelAreaTask extends BaseTask[ChannelArea]{ 
      override def map(clickLogWideDataStream: DataStream[ClickLogWide]): DataStream[ChannelArea] = { 
        val isOld = (isNew:Int,isDataNew:Int)=> if (isNew == 0 && isDataNew ==1) 1 else 0
        clickLogWideDataStream.flatMap{ 
          clickLogWide=>
            List(
              ChannelArea(clickLogWide.channelID,
                clickLogWide.address,
                clickLogWide.yearMonth,
                clickLogWide.count,
                clickLogWide.isMonthNew,
                clickLogWide.isNew,
                isOld(clickLogWide.isNew,clickLogWide.isMonthNew) ),
              ChannelArea(clickLogWide.channelID,
                clickLogWide.address,
                clickLogWide.yearMonthDay,
                clickLogWide.count,
                clickLogWide.isDayNew,
                clickLogWide.isNew,
                isOld(clickLogWide.isNew, clickLogWide.isDayNew)
              ),
              ChannelArea(clickLogWide.channelID,
                clickLogWide.address,
                clickLogWide.yearMonthDayHour,
                clickLogWide.count,
                clickLogWide.isHourNew,
                clickLogWide.isNew,
                isOld(clickLogWide.isNew, clickLogWide.isHourNew)
              )
            )
    
        }
      }
    
      override def keyBy(mapDataStream: DataStream[ChannelArea]): KeyedStream[ChannelArea, String] = { 
        mapDataStream.keyBy{ 
          area=> (area.channelId + ":" + area.area + ":" +area.date)
        }
      }
    
      override def timeWindow(keyedStream: KeyedStream[ChannelArea, String]): WindowedStream[ChannelArea, String, TimeWindow] = { 
        keyedStream.timeWindow(Time.seconds(3))
      }
    
      override def reduce(windowedStream: WindowedStream[ChannelArea, String, TimeWindow]): DataStream[ChannelArea] = { 
        windowedStream.reduce{ 
          (t1,t2)=>
            ChannelArea(t1.channelId,t1.area,
              t1.date,
              t1.pv+t2.pv,
              t1.uv+t2.uv,
              t1.newCount+t2.newCount,
              t1.oldCount+t2.oldCount)
        }
      }
    
      override def sink2HBase(reduceDataStream: DataStream[ChannelArea]): Unit = { 
        reduceDataStream.addSink{ 
          area=>{ 
            //HBase相关字段
            val tableName = "channel_area"
            val clfName = "info"
            val rowkey = area.channelId + ":" + area.area + ":" +area.date
    
            val channelIdColumn = "channelId"
            val areaColumn = "area"
            val dateColumn = "date"
            val pvColumn = "pv"
            val uvColumn = "uv"
            val newCountColumn = "newCount"
            val oldCountColumn = "oldCount"
    
            //查询hbase
            val pvInHbase: String = HBaseUtil.getData(tableName,rowkey,clfName,pvColumn)
            val uvInHbase: String = HBaseUtil.getData(tableName,rowkey,clfName,uvColumn)
            val newCountInHbase: String = HBaseUtil.getData(tableName,rowkey,clfName,newCountColumn)
            val oldCountInHbase: String = HBaseUtil.getData(tableName,rowkey,clfName,oldCountColumn)
    
            //累加
            var totalPv = 0L
            var totalUv = 0L
            var totalNewCount = 0L
            var totalOldCount = 0L
    
            if (StringUtils.isNotBlank(pvInHbase)){ 
              totalPv = pvInHbase.toLong + area.pv
            }else{ 
              totalPv = area.pv
            }
    
            if(StringUtils.isNotBlank(uvInHbase)){ 
              totalUv  = uvInHbase.toLong+area.uv
            }else{ 
              totalUv = area.uv
            }
    
            if(StringUtils.isNotBlank(newCountInHbase)){ 
              totalNewCount  = newCountInHbase.toLong+area.newCount
            }else{ 
              totalNewCount = area.newCount
            }
    
            if(StringUtils.isNotBlank(oldCountInHbase)){ 
              totalOldCount  = oldCountInHbase.toLong+area.oldCount
            }else{ 
              totalOldCount = area.oldCount
            }
    
            //保存数据
            HBaseUtil.putMapData(tableName,rowkey,clfName,Map(
              channelIdColumn->area.channelId,
              areaColumn->area.area,
              dateColumn->area.date,
              pvColumn->totalPv,
              uvColumn->totalUv,
              newCountColumn->totalNewCount,
              oldCountColumn->totalOldCount
            ))
          }
        }
      }
    }

# 实时运营商分析业务开发 #

## 业务介绍 ##

分析出来中国移动、中国联通、中国电信等运营商的指标。来分析，流量的主要来源是哪个运营商的，这样就可以进行较准确的网络推广。  
**需要分析出来指标**

*  PV
 *  UV
 *  新用户
 *  老用户  
    **需要分析的维度**
 *  运营商
 *  时间维度（时、天、月）  
    统计分析后的结果如下：

## 业务开发 ##

**步骤**

1.  将预处理后的数据，转换为要分析出来数据（频道、运营商、时间、PV、UV、新用户、老用户）样例类
2.  按照`频道`、`时间`、`运营商`进行分组（分流）
3.  划分时间窗口（3秒一个窗口）
4.  进行合并计数统计
5.  打印测试
6.  将计算后的数据下沉到Hbase  
    **实现**
7.  创建一个`ChannelNetworkTask`单例对象
8.  添加一个`ChannelNetwork`样例类，它封装要统计的四个业务字段：频道ID（channelID）、运营商（network）、日期（date）pv、uv、新用户（newCount）、老用户（oldCount）
9.  在`ChannelNetworkTask`中编写一个`process`方法，接收预处理后的`DataStream`
10. 使用`flatMap`算子，将`ClickLog`对象转换为三个不同时间维度`ChannelNetwork`
11. 按照`频道ID`、`时间`、`运营商`进行分流
12. 划分时间窗口（3秒一个窗口）
13. 执行reduce合并计算
14. 打印测试
15. 将合并后的数据下沉到hbase
    
     *  准备hbase的表名、列族名、rowkey名、列名
     *  判断hbase中是否已经存在结果记录
     *  若存在，则获取后进行累加
     *  若不存在，则直接写入  
        **参考代码**

import com.itheima.realprocess.bean.ClickLogWide
    import com.itheima.realprocess.util.HBaseUtil
    import org.apache.commons.lang.StringUtils
    import org.apache.flink.streaming.api.scala.{ DataStream, KeyedStream, WindowedStream}
    import org.apache.flink.streaming.api.windowing.windows.TimeWindow
    import org.apache.flink.api.scala._
    
    case class ChannelNetWork(
                               var channelId: String,
                               var network: String,
                               var date: String,
                               var pv: Long,
                               var uv: Long,
                               var newCount: Long,
                               var oldCount: Long
                             )
    
    object ChannelNetWorkTask extends BaseTask[ChannelNetWork] { 
      override def map(clickLogWideDataStream: DataStream[ClickLogWide]): DataStream[ChannelNetWork] = { 
    
        val isOld = (isNew: Int, isDateNew: Int) => if (isNew == 0 && isDateNew == 1) 1 else 0
        clickLogWideDataStream.flatMap { 
          clickLogWide => { 
            List(
              ChannelNetWork(
                clickLogWide.channelID,
                clickLogWide.network,
                clickLogWide.yearMonth,
                clickLogWide.count,
                clickLogWide.isMonthNew,
                clickLogWide.isNew,
                isOld(clickLogWide.isNew, clickLogWide.isMonthNew)
              ),
              ChannelNetWork(
                clickLogWide.channelID,
                clickLogWide.network,
                clickLogWide.yearMonthDay,
                clickLogWide.count,
                clickLogWide.isDayNew,
                clickLogWide.isNew,
                isOld(clickLogWide.isNew, clickLogWide.isDayNew)
              ),
              ChannelNetWork(
                clickLogWide.channelID,
                clickLogWide.network,
                clickLogWide.yearMonthDayHour,
                clickLogWide.count,
                clickLogWide.isHourNew,
                clickLogWide.isNew,
                isOld(clickLogWide.isNew, clickLogWide.isHourNew)
              )
            )
          }
        }
      }
    
      override def keyBy(mapDataStream: DataStream[ChannelNetWork]): KeyedStream[ChannelNetWork, String] = { 
        mapDataStream.keyBy { 
          network =>
            network.channelId + ":" + network.network + ":" + network.date
        }
      }
    
      override def reduce(windowedStream: WindowedStream[ChannelNetWork, String, TimeWindow]): DataStream[ChannelNetWork] = { 
        windowedStream.reduce { 
          (t1, t2) => { 
            ChannelNetWork(t1.channelId, t1.network, t1.date,
              t1.pv + t2.pv, t1.uv + t2.uv, t1.newCount + t2.newCount, t1.oldCount + t2.oldCount)
          }
        }
      }
    
      override def sink2HBase(reduceDataStream: DataStream[ChannelNetWork]): Unit = { 
    
        reduceDataStream.addSink(
          network => { 
            // 创建HBase相关列
            val tableName = "channel_network"
            val clfName = "info"
            val rowkey = s"${network.channelId}:${network.network}:${network.date}"
    
            val channelIdColumn = "channelId"
            val networkColumn = "network"
            val dateColumn = "date"
            val pvColumn = "pv"
            val uvColumn = "uv"
            val newCountColumn = "newCount"
            val oldCountColumn = "oldCount"
    
            // 查询HBase
    
            val resultMap: Map[String, String] = HBaseUtil.getMapData(tableName, rowkey, clfName, List(
              pvColumn, uvColumn, newCountColumn, oldCountColumn
            ))
    
            // 数据累加
    
            var totalPv = 0L
            var totalUv = 0L
            var totalNewCount = 0L
            var totalOldCount = 0L
    
            // 如果resultMap不为空,并且可以去到相关列的值,那么就进行累加
            if (resultMap != null && StringUtils.isNotBlank(resultMap.getOrElse(pvColumn,""))) { 
              totalPv = resultMap(pvColumn).toLong + network.pv
            } else { 
              totalPv = network.pv
            }
    
    
            if (resultMap != null && StringUtils.isNotBlank(resultMap.getOrElse(uvColumn,""))) { 
              totalUv = resultMap(uvColumn).toLong + network.uv
            } else { 
              totalUv = network.uv
            }
    
            if (resultMap != null && StringUtils.isNotBlank(resultMap.getOrElse(newCountColumn,""))) { 
              totalNewCount = resultMap(newCountColumn).toLong + network.newCount
            } else { 
              totalNewCount = network.newCount
            }
    
            if (resultMap != null && StringUtils.isNotBlank(resultMap.getOrElse(oldCountColumn,""))) { 
              totalOldCount = resultMap(oldCountColumn).toLong + network.oldCount
            } else { 
              totalOldCount = network.oldCount
            }
    
    
            // 保存数据
            HBaseUtil.putMapData(
              tableName, rowkey, clfName, Map(
                channelIdColumn -> network.channelId,
                networkColumn -> network.network,
                dateColumn -> network.date,
                pvColumn -> totalPv,
                uvColumn -> totalUv,
                newCountColumn -> totalNewCount,
                oldCountColumn -> totalOldCount
              )
            )
          }
        )
    
      }
    }

# 实时频道浏览器分析业务开发 #

## 业务介绍 ##

需要分别统计不同浏览器（或者客户端）的占比  
**需要分析出来指标**

*  PV
 *  UV
 *  新用户
 *  老用户  
    **需要分析的维度**
 *  浏览器
 *  时间维度（时、天、月）  
    统计分析后的结果如下：

<table> 
 <thead> 
  <tr> 
   <th>频道ID</th> 
   <th>浏览器</th> 
   <th>时间</th> 
   <th>PV</th> 
   <th>UV</th> 
   <th>新用户</th> 
   <th>老用户</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>频道1</td> 
   <td>360浏览器</td> 
   <td>201809</td> 
   <td>1000</td> 
   <td>300</td> 
   <td>0</td> 
   <td>300</td> 
  </tr> 
  <tr> 
   <td>频道1</td> 
   <td>IE</td> 
   <td>20180910</td> 
   <td>123</td> 
   <td>1</td> 
   <td>0</td> 
   <td>1</td> 
  </tr> 
  <tr> 
   <td>频道1</td> 
   <td>Chrome</td> 
   <td>2018091010</td> 
   <td>55</td> 
   <td>2</td> 
   <td>2</td> 
   <td>0</td> 
  </tr> 
 </tbody> 
</table>

## 业务开发 ##

**步骤**

1.  创建频道浏览器分析样例类（频道、浏览器、时间、PV、UV、新用户、老用户）
2.  将预处理后的数据，使用`flatMap`转换为要分析出来数据样例类
3.  按照`频道`、`时间`、`浏览器`进行分组（分流）
4.  划分时间窗口（3秒一个窗口）
5.  进行合并计数统计
6.  打印测试
7.  将计算后的数据下沉到Hbase  
    **实现**
8.  创建一个`ChannelBrowserTask`单例对象
9.  添加一个`ChannelBrowser`样例类，它封装要统计的四个业务字段：频道ID（channelID）、浏览器（browser）、日期（date）pv、uv、新用户（newCount）、老用户（oldCount）
10. 在`ChannelBrowserTask`中编写一个`process`方法，接收预处理后的`DataStream`
11. 使用`flatMap`算子，将`ClickLog`对象转换为三个不同时间维度`ChannelBrowser`
12. 按照`频道ID`、`时间`、`浏览器`进行分流
13. 划分时间窗口（3秒一个窗口）
14. 执行reduce合并计算
15. 打印测试
16. 将合并后的数据下沉到hbase
    
     *  准备hbase的表名、列族名、rowkey名、列名
     *  判断hbase中是否已经存在结果记录
     *  若存在，则获取后进行累加
     *  若不存在，则直接写入  
        **参考代码**

import com.itheima.realprocess.bean.ClickLogWide
    import com.itheima.realprocess.util.HBaseUtil
    import org.apache.commons.lang.StringUtils
    import org.apache.flink.streaming.api.scala.{ DataStream, KeyedStream, WindowedStream}
    import org.apache.flink.streaming.api.windowing.windows.TimeWindow
    import org.apache.flink.api.scala._
    import org.apache.flink.streaming.api.windowing.time.Time
    
    case class ChannelBrowser(var channelID: String,
                              var browser: String,
                              var date: String,
                              var pv: Long,
                              var uv: Long,
                              var newCount: Long,
                              var oldCount: Long
                             )
    
    object ChannelBrowserTask extends BaseTask[ChannelBrowser]{ 
      override def map(clickLogWideDataStream: DataStream[ClickLogWide]): DataStream[ChannelBrowser] = { 
    
        val isOld = (isNew: Int, isDateNew: Int) => if (isNew == 0 && isDateNew == 1) 1 else 0
    
        clickLogWideDataStream.flatMap{ 
          clickLogWide =>{ 
            List (
              ChannelBrowser(
              clickLogWide.channelID,
              clickLogWide.browserType,
              clickLogWide.yearMonth,
              clickLogWide.count,
              clickLogWide.isMonthNew,
              clickLogWide.isNew,
              isOld(clickLogWide.isNew, clickLogWide.isMonthNew)
            ),
            ChannelBrowser(
              clickLogWide.channelID,
              clickLogWide.browserType,
              clickLogWide.yearMonthDay,
              clickLogWide.count,
              clickLogWide.isDayNew,
              clickLogWide.isNew,
              isOld(clickLogWide.isNew, clickLogWide.isDayNew)
            ),
            ChannelBrowser(
              clickLogWide.channelID,
              clickLogWide.browserType,
              clickLogWide.yearMonthDayHour,
              clickLogWide.count,
              clickLogWide.isHourNew,
              clickLogWide.isNew,
              isOld(clickLogWide.isNew, clickLogWide.isHourNew)
            )
            )
          }
        }
      }
    
      override def keyBy(mapDataStream: DataStream[ChannelBrowser]): KeyedStream[ChannelBrowser, String] = { 
         mapDataStream.keyBy{ 
           browser =>
             browser.channelID + ":" + browser.browser + ":" + browser.date
        }
      }
    
      override def timeWindow(keyedStream: KeyedStream[ChannelBrowser, String]): WindowedStream[ChannelBrowser, String,
        TimeWindow] = { 
        keyedStream.timeWindow(Time.seconds(3))
      }
    
      override def reduce(windowedStream: WindowedStream[ChannelBrowser, String, TimeWindow]): DataStream[ChannelBrowser] = { 
    
        windowedStream.reduce { 
          (t1, t2) => { 
            ChannelBrowser(t1.channelID,t1.date,t1.date,
              t1.pv+t2.pv,
              t1.uv+t2.uv,
              t1.newCount+t2.newCount,
              t1.oldCount+t2.oldCount)
          }
        }
      }
    
      override def sink2HBase(reduceDataStream: DataStream[ChannelBrowser]): Unit = { 
        reduceDataStream.addSink(
          browser=>{ 
            //创建HBase相关列、
            val tableName = "channel_browser"
            val clfName = "info"
            val rowkey = s"${browser.channelID}:${browser.browser}:${browser.date}"
    
            val channelIDColumn = "channelID"
            val browserColumn = "browser"
            val dateColumn = "date"
            val pvColumn = "pv"
            val uvColumn = "uv"
            val newCountColumn = "newCount"
            val oldCountColumn = "oldCount"
    
            //查询Hbase
            val resultMap = HBaseUtil.getMapData(tableName, rowkey, clfName, List(
              pvColumn, uvColumn, newCountColumn, oldCountColumn
            ))
    
            var totalPv = 0L
            var totalUv = 0L
            var totalNewCount = 0L
            var totalOldCount = 0L
    
    
            //如果resultMap不为空，并且可以去到相关列的值，那么就进行累加
            if(resultMap != null && StringUtils.isNotBlank(resultMap.getOrElse(pvColumn,""))){ 
              totalPv = resultMap(pvColumn).toLong+browser.pv
            }else{ 
              totalPv = browser.pv
            }
    
            if (resultMap != null && StringUtils.isNotBlank(resultMap.getOrElse(pvColumn,""))){ 
              totalUv = resultMap(uvColumn).toLong + browser.uv
            }else{ 
              totalUv = browser.uv
            }
    
            if (resultMap != null && StringUtils.isNotBlank(resultMap.getOrElse(newCountColumn,""))){ 
              totalNewCount = resultMap(newCountColumn).toLong + browser.newCount
            }else{ 
              totalNewCount = browser.newCount
            }
    
            if (resultMap != null && StringUtils.isNotBlank(resultMap.getOrElse(oldCountColumn,""))){ 
              totalOldCount = resultMap(oldCountColumn).toLong + browser.oldCount
            }else{ 
              totalOldCount = browser.oldCount
            }
    
            //保存数据到HBase
            HBaseUtil.putMapData(
              tableName,rowkey,clfName,Map(
                  channelIDColumn -> browser.channelID,
                  browserColumn -> browser.browser,
                  dateColumn -> browser.date,
                  pvColumn -> browser.pv,
                  uvColumn -> browser.uv,
                  newCountColumn -> browser.newCount,
                  oldCountColumn -> browser.oldCount
              )
            )
          }
        )
      }
    }

# 数据输不出来请看这 #

低级错误  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70 1]  
请改成  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70 2]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70]: /images/20221023/049f84640b84446c8e5df0f0deba24f4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70 1]: /images/20221023/604541e83782473798fbee2ce233c652.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70 2]: /images/20221023/9da71f480f4d4caa9065ca042a23f0aa.png