实战
数据导入Hive中
全量： 拉链增量：用户、商品表数据量大时用 拉链表动作表 增量城市信息 全量需求一： 获取点击、下单和支付数量排名前 10 的品类①使用累加器：click_category_id，个数order_category_ids，个数 pay_category_ids，个数 ②在Driver端进行累加处理click_category_id_click，个数order_category_ids_order，个数pay_category_ids_pay，个数(9_click,2977)(2_click,2898)(1,2,3_order,17922)(1,2,3_pay,11802)③分组合并groupBy(click_category_id，Map(click_category_id_click -> 个数))(order_category_ids，Map(order_category_ids_order -> 个数))(pay_category_ids, Map(order_category_ids_order -> 个数))(12,Map(12_click -> 2982))(1,2,3,Map(1,2,3_order -> 17922, 1,2,3_pay -> 11802))(3,Map(3_click -> 3032))④将合并的数据转换为一条数据mapCategoryTop10(taskid, categoryid, clickcount, ordercount, paycount)CategoryTop10(13482102-852b-4403-99d2-1ebb99292df6,8,2988,0,0) ⑤把它变成集合可排序，将数据排序后取前10条 .sortWith .take(10)按clickcount排序，从大到小CategoryTop10(7a867338-d4b6-4da4-b7c0-fbd7c1477157,11,3050,0,0)CategoryTop10(7a867338-d4b6-4da4-b7c0-fbd7c1477157,9,2977,0,0)
一个用户可能会有多个session，所以不按用户进行分组而是按session
需求二： Top10 热门品类中 Top10 活跃 Session 统计① 获取前10的品类数据 CategoryTop10(taskid, categoryid, clickcount, ordercount, paycount)②将当前的日志数据进行筛选过滤filter（品类，点击）筛选出categoryid, clickcountcategoryIds <-- top10Data.map(_.categoryId)从原始数据中过滤出top10的品类: click_category_id != -1 categoryIds.contains("" + action.click_category_id)UserVisitAction(2019-11-07,19,bf9dfe31-8703-49a4-a513-78ce49758d08,11,2019-11-07 16:13:24,null,18,31,null,null,null,null,18)③对筛选后的数据进行聚合：filterRDD.map( ).reduceByKey(_+_) 转换结构，聚合(categoryId_sessionId, clickCount)(8_fae66028-0fe7-44f7-9720-e24665523e9e,1)(12_1b1328cb-fabd-4056-a029-416d9c7e6c78,3)④根据品类进行分组.map split("_") (categoryId, (sessionId, clickCount)) (12,(1b1328cb-fabd-4056-a029-416d9c7e6c78,3)).groupBy 只保留V即可groupBy会把分组之后的保存在CompactBuffer中((3315e7ec-931f-45a0-95f2-183208f9ca34,1),CompactBuffer((15,(3315e7ec-931f-45a0-95f2-183208f9ca34,1)), (18,(3315e7ec-931f-45a0-95f2-183208f9ca34,1))))⑤变成List再排序.sortWith（降序），获取前10条数据.mapValues() 是针对于(K,V)形式的类型只对V进行操作排序.sortWith--scala中的一个排序方法，而sortBy是可以自定义排序规则 List((12,(3196a615-f034-4a6b-a3b2-aab8a928addc,1)), (11,(3196a615-f034-4a6b-a3b2-aab8a928addc,1))) 针对V排序 ._2._2对clickCount排序从高到低 转换结构map(_._2) 只保留V即可--> List((18,(ffe3e64f-7e0a-40a0-9b63-d0264e00de65,1)), (16,(ffe3e64f-7e0a-40a0-9b63-d0264e00de65,1)))⑥偏平化处理变成一条一条.flatMap(x => x)(categoryId, (sessionId, clickCount))(18,(ffe3e64f-7e0a-40a0-9b63-d0264e00de65,1))(16,(ffe3e64f-7e0a-40a0-9b63-d0264e00de65,1))
需求3 页面单跳转化率统计 1. 获取分母：每一个页面点击次数总和 ①从配置文件中读取并切分.split转成Array类型 pageids ②过滤,从原始数据(.contains)中过滤出符合 pageid的数据 ③转换结构(pageid, 1L)如: (2,1) (5,1)并聚合.reduceByKey(_+_)统计 (2,1807) (5,1777) ④转换结构为map类型k, v对的形式 (5,1777), (1,1804)2. 获取分子：应该是根据过滤之前的数据进行统计，不应该是过滤之后的数据 ① 根据session进行分组， 不应该根据用户分组，一个用户可以访问多个 .groupBy 按传入函数的返回值 session_id进行分组 (cff267a0-36f5-4be6-abe1-51bcaee5e976,CompactBuffer(UserVisitAction(2019-06-06,3,cff267a0-36f5-4be6-abe1-51bcaee5e976,29,2019-06-06 09:56:35,null,3,3,null,null,null,null,3), UserVisitAction(2019-06-06,3,cff267a0-36f5-4be6-abe1-51bcaee5e976,47,2019-06-06 09:57:37,i7,-1,-1,null,null,null,null,10))) ②将分组后的数据进行排序 .mapValues{ .toList.sortWith } //mapValues是对V进行操作排序，按action_time从大到小排 .map(_.page_id) .zip(pageidList.tail) 再使用拉链 zip 将List(1,2,5,4, 6) zip List(2,5,4, 6) 组合在一块(1-2), (2-5), (5-4), (4-6)； 再转换结构.map ( (1-2, 1), (2-5, 1), (5-4, 1), (4-6, 1) ) ===>>(437ffda2-0e5b-4374-b5ca-b86441b10dc9,List((8-4,1), (4-3,1), (3-16,1), (16-33,1))) ③转换结构,只保留V List类型即可List((8-4,1), (4-3,1), (3-16,1), (16-33,1)) 再扁平化 .flatMap(x => x) -->(8-4,1) (4-3,1) (3-16,1) (16-33,1) ④将不需要关心页面流转的数据过滤掉 .filter (pageids.zip(pageids.tail).map()).contains(k) k即上边数据(8-4,1)中的k即8-4 pageids.zip(pageids.tail).map() 1-2 2-3 3-4 4-5 5-6 6-7 ⑤统计符合要求的个数.reduceByKey(_+_) // (1-2, 100)3. 页面单跳点击次数 / 页面点击次数 遍历分子(1-2, 100)， 分母： (5,1777), (1,1804) 切分split("-")取出第一个(0)作为key值取出value即分母 补充需求：网站页面平均停留时长
网站页面平均停留时长1 获取离线数据2 将数据根据session进行分组 .groupBy3 将分组后的数据按照访问时间进行排序 .mapValues(datas => { datas.toList.sortWith { }.map() //(page_id,action_time).zip( .tail).map() //(before._1, (endTime - startTime)) })4 采用拉链的方式将数据进行组合，形成页面跳转路径（A=>B,B=>C） .map() //不关心session了 将集合数据进行扁平化操作.flatMap(x => x)5 对跳转路径中的第一个页面进行分组.groupByKey()，聚合数据 .mapValues(datas => { datas.sum / datas.size })6 对聚合后的数据进行计算，获取结果.foreach(println)
实时数据分析
redis
redis中5大数据类型是指V，K都是String①Set:set对外提供的功能与list类似是一个列表的功能，特殊之处在于set是可以自动排重的，当你需要存储一个列表数据，又不希望出现重复数据时，set是一个很好的选择，并且set提供了判断某个成员是否在一个set集合内的重要接口，这个也是list所不能提供的。Redis的Set是string类型的无序集合。它底层其实是一个value为null的hash表,所以添加，删除，②hash 是一个键值对集合。Redis hash是一个string类型的field和value的映射表，hash特别适合用于存储对象。类似Java里面的Map
DSream 代表了一系列连续的RDD，DStream中每个RDD包含特定时间间隔的数据
实时数据分析： 广告黑名单实时统计
实现实时的动态黑名单机制：将每天对某个广告点击超过 100 次的用户拉黑。
注：黑名单保存到redis中。
已加入黑名单的用户不在进行检查。
[kris@hadoop101 bin]$ ./kafka-topics.sh --zookeeper hadoop101:2181 --list
kafka中创建topic: ads_log
[kris@hadoop101 bin]$ ./kafka-topics.sh --zookeeper hadoop101:2181,hadoop102:2181,hadoop103:2181 --create --topic ads_log --partitions 3 --replication-factor 3[kris@hadoop101 bin]$ ./kafka-console-consumer.sh --zookeeper hadoop101:2181 --topic ads_log --from-beginning
序列化、反序列化--生产者、消费者
kafka分段日志，spark分段文件；序列化调用writeReplace()
需求4：广告黑名单实时统计kafka中的数据格式：timestamp province city userid adid1555928451187 华北 北京 4 21555928451187 华北 北京 3 6
1.从kafka中周期性获取广告点击数据KafkaUtil.getKafkaStream(topic, streamContext)2.将数据进行分解和转换： ①转换成含有这样类型的kafkaMessage的数据 .map()，使用样例类 ②将获取的数据进行筛选（过滤掉黑名单的数据）使用.transform .transform(rdd => { ②.1获取Jedis的连接； ②.2取出集合中的所有值，使用redis的Set类型 ②.3设置广播变量(可实现序列化和向所有节点发送数值) 这些都是在Driver端执行，可随着executor执行相同的次数 rdd.filter({ 在executor端执行 }) }) ②.4关闭jedis.close() 另外一种方法是将cp状态值保存到redis中，这样就不用下面这些步骤了；
.foreachRDD(rdd=>{ rdd.foreachPartition(messages=>{ for (message <- messages) { val dateString: String = SparkmallUtils.parseStringDateFromTs(message.ts.toLong, "yyyy-MM-dd") val key = "date:user:ad:clickcount" //dateString + "_" + message.userid + "_" + message.adid val field = dateString + "_" + message.userid + "_" + message.adid jedis.hincrBy(key, field, 1) // 将redis中指定的key集合，对field字段的值进行数据累加；选择的redis类型为hash val sumString: String = jedis.hget(key, field) //获取值进行判断是否点击次数超过100次，超过就添加到黑名单中 val sum = sumString.toLong if ( sum >= 100 ) { jedis.sadd("blacklist", message.userid) //将黑名单也添加到redis集合中，类型为set } } jedis.close() } }) }) ③转换为这种格式:（ts_user_ad, 1）3 将转换的结果进行聚合 （采集周期内）：（ts_user_ad, sum）4 将不同周期中采集的数据进行累加（有状态的） ①设定CP的路径 streamingContext.sparkContext.setCheckpointDir("cp") ②.updateStateByKey{ case (seq, opt) => { //Seq[Int], Option[s] Seq可直接.sum // 将当前采集周期的统计结果和CP中的数据进行累加 val sum: Int = seq.sum + opt.getOrElse(0) Option(sum) // 将累加的结果更新到CP中 } }5 将累加的结果进行判断，是否超过阈值（100） .foreachRDD(rdd => { rdd.foreach{ case(key, sum) => { if(sum > 100){6 如果超过阈值，那么将用户加入黑名单，防止用户继续访问 val jedis: Jedis = new Jedis("hadoop101", 6379) val userid: String = key.split("_")(1)// 获取用户 jedis.sadd("blacklist", userid) } } } })127.0.0.1:6379> SMEMBERS blacklist //foreachRDD参数应该实现将每一个RDD中数据推送到外部系统，如将RDD存入文件或者通过网络将其写入数据库。
//对DStream中的RDD采取任意的操作需求5： 广告点击量实时统计
1.从kafka中周期性获取广告点击数据KafkaUtil.getKafkaStream(topic, streamContext)2.将数据进行分解和转换： ①转换成含有这样类型的kafkaMessage的数据 .map()，使用样例类 方法二：可直接添加到redis中，省去下面的步骤了； .foreachRDD(rdd => { rdd.foreachPartition(messages => { val jedis: Jedis = RedisUtil.getJedisClient for (message <- messages) { val dateString: String = SparkmallUtils.parseStringDateFromTs(message.ts.toLong, "yyyy-MM-dd") val key = "date:area:city:ads" val field = dateString + "_" + message.area + "_" + message.city + "_" + message.adid jedis.hincrBy(key, field, 1) } jedis.close() }) }) ②将采集周期中的数据进行聚合转换结构--> (ds_area_city_ad, 1)3.将不同周期中采集的数据进行累加（有状态的） ①设定CP的路径 streamingContext.sparkContext.setCheckpointDir("cp") ②.updateStateByKey{ case (seq, opt) => { //Seq[Int], Option[s] Seq可直接.sum // 将当前采集周期的统计结果和CP中的数据进行累加 val sum: Int = seq.sum + opt.getOrElse(0) Option(sum) // 将累加的结果更新到CP中 } }(2019-04-22_华北_北京_1,109)(2019-04-22_华北_天津_2,40)(2019-04-22_华东_上海_6,133) 4.将聚合后的数据更新到redis中； 是更新不是累加，跟上一个需求有点不一样；.foreachRDD(rdd => { // RDD[(String, Int )] => Unit rdd.foreachPartition(datas => { //foreachPartition用于在每个分区创建一个连接 val jedis: Jedis = RedisUtil.getJedisClient for ((field, sum) <- datas) { val key = "date:area:city:ads" jedis.hset(key, field, "" +sum) } jedis.close() }) }) 基于需求5广告点击量实时统计 --> 需求6 每天各地区 top3 热门广告
每天各地区 top3 热门广告1 获取需求5的数据2 将获取的数据进行格式转换（date_area_city_adv,sum）-->(date_area_adv,sum) 转换结构.map 3 将转换的数据进行聚合统计：(date_area_adv,sum)--> (date_area_adv,totalSum) .reduceByKey(_+_) 4 将统计的结果进行结构转换：(date_area_adv,totalSum) -->( date_area, ( adv, totalSum ) ) .map5 将转换后的数据进行分组排序：（ date_area，Map( adv1totalSum1, adv2totalSum2 ) ） .groupByKey()(2019-04-22_华东,ArrayBuffer((上海,199)))(2019-04-22_华南,ArrayBuffer((广州,49), (深圳,118)))(2019-04-22_华北,ArrayBuffer((天津,57), (北京,138))) .mapValues(datas => { datas.toList.sortWith { case (left, right) => { left._2 > right._2 } }.take(3).toMap //对数据进行排序，获取排序后数据的前三名; 转换成map集合类型 })6 将结果保存到redis中.foreachRDD(rdd => { rdd.foreachPartition(datas => { val jedis: Jedis = RedisUtil.getJedisClient for ((k, map) <- datas) { val ks: Array[String] = k.split("_") val key = "top3_ads_per_day:" + ks(0) // 将Scala集合转换为JSON字符串；因为结果是{ }类型的 import org.json4s.JsonDSL._ val value: String = JsonMethods.compact(JsonMethods.render(map)) jedis.hset(key, ks(1), value) } jedis.close() }) }) 需求7：最近一小时广告点击趋势
1.从kafka中周期性获取广告点击数据KafkaUtil.getKafkaStream(topic, streamContext)2.将数据进行分解和转换： ①转换成含有这样类型的kafkaMessage的数据 .map()，使用样例类3.使用窗口函数将多个采集周期的数据作为一个整体进行统计 .window(Seconds(60), Seconds(10)) //周期60s，步长10s4 获取数据，将数据根据窗口的滑动的幅度进行分组 将数据进行结构的转换（KafkaMessage）==> (time,1) .map(message => { var time: String = SparkmallUtils.parseStringDateFromTs(message.ts.toLong, "yyyy-MM-dd HH:mm:ss") val prefixTime: String = time.substring(0, time.length - 1) time = prefixTime + "0" //将秒变成的个位数去掉，补0 (time, 1) })5 将分组后的数据进行统计聚合.reduceByKey(_ + _)6 将统计的结果按照时间进行排序.transform(rdd => { //转换，用时间进行排序,false是从小往大 rdd.sortBy(t => { t._1 }, false) })