hadoop大数据Flink专题篇-Flink四种API入门实战（二）

package com.zhenglihan.cdh.flink


import java.text.SimpleDateFormat
import java.util.{Calendar}


import org.apache.flink.api.common.functions.GroupReduceFunction
import org.apache.flink.api.scala._
import org.apache.flink.util.Collector


import scala.collection.JavaConversions
/**
  * 功能描述：
  *   统计“一小时内”用户的访问链接次数（注意理解一小时内的含义）
  * 实现方案：
  *   采用flink Dataset API 加自定义函数
  * 关键知识点：
  *   GroupReduceFunction
  * 输入：
  *   DataSet
  * 输出：
  *   控制台
  * 输出结果：
  *   (Bob,2018-12-22 12:00:00,1)
  *   (Bob,2018-12-22 13:01:00,1)
  *   (Bob,2018-12-22 14:30:00,1)
  *   (Bob,2018-12-22 14:40:00,2)
  *   (Bob,2018-12-22 15:40:00,1)
  *   (Liz,2018-12-22 13:30:00,1)
  *   (Liz,2018-12-22 13:59:00,2)
  *   (Liz,2018-12-22 14:02:00,3)
  *   (Mary,2018-12-22 12:00:00,1)
  *   (Mary,2018-12-22 12:02:00,2)
  *   (Mary,2018-12-22 12:55:00,3)
  *   (Mary,2018-12-22 14:00:00,1)
  */
object FlinkBatchDataSetUserClickDemo {


  def main(args: Array[String]): Unit = {


    // set up execution environment
    val env = ExecutionEnvironment.getExecutionEnvironment


    val input = env.fromElements(
      UserClick("Mary","2018-12-22 12:00:00","./home"),
      UserClick("Bob","2018-12-22 12:00:00","./cart"),
      UserClick("Mary","2018-12-22 12:02:00","./prod?id=1"),
      UserClick("Mary","2018-12-22 12:55:00","./prod?id=4"),
      UserClick("Bob","2018-12-22 13:01:00","./prod?id=5"),
      UserClick("Liz","2018-12-22 13:30:00","./home"),
      UserClick("Liz","2018-12-22 13:59:00","./prod?id=7"),
      UserClick("Mary","2018-12-22 14:00:00","./cart"),
      UserClick("Liz","2018-12-22 14:02:00","./home"),
      UserClick("Bob","2018-12-22 14:30:00","./prod?id=3"),
      UserClick("Bob","2018-12-22 14:40:00","./home"),
      UserClick("Bob","2018-12-22 15:40:00","./home")
    )
    input.groupBy(_.user).reduceGroup(new UserClickGroupReduceFunction).print()


  }


  case class UserClick(user: String, time: String, url:String)
  class UserClickGroupReduceFunction extends GroupReduceFunction[UserClick, (String,String,Long)] {
    override def reduce(input: java.lang.Iterable[UserClick], out: Collector[(String,String,Long)]) = {
      val list=JavaConversions.iterableAsScalaIterable(input).toList.sortBy(_.time)
      list.foreach(userClick=>{
        var count=0
        for(i<-0 to list.size-1){
          val userClick2=list(i)
          //统计list中userClick.time数据时间在一小时以内的数据
          if(userClick2.time.compareTo(getOneHourBefore(userClick.time,-1)+userClick.time.substring(13))>0 && userClick2.time.compareTo(userClick.time)<=0){
            count=count+1
          }
        }
        out.collect((userClick.user,userClick.time,count))
      })
      out
    }


    /**
      * 获取preHour小时后的时间，精确到小时，例如输入：2018-12-22 13:59:00,-1 输出：2018-12-22 12
      * @param time
      * @param preHour
      * @return
      */
    def getOneHourBefore(time:String,preHour:Int): String = {
      val cal = Calendar.getInstance
      val sdf = new SimpleDateFormat("yyyy-MM-dd HH");
      cal.setTime(sdf.parse(time))
      cal.add(Calendar.HOUR_OF_DAY, preHour)
      val newTime = sdf.format(cal.getTime)
      newTime
    }
  }
}