3000字 | 一文讲透redis在大数据开发中的应用场景

hset 1 age 10
hset 2 age 30
hset 3 age 14

redis-cli [options]  < filename

val conf = new SparkConf().setAppName("test").setMaster("local[*]")
val sc = new SparkContext(conf)
if (args.length != 5) {
  print("command arguments less than 5，please check again")
  sc.stop()
}
val lines = sc.textFile(args(0))
// 0 文件路径 1 IP 2 Port 3 password 4 操作
val clusterPipeline = sc.broadcast(JedisClusterPool(args(1), args(2).toInt, args(3), -1, -1, 3, 1000, 1000))
val operator = args(4)
lines
  .map(_.split(" "))
  .foreachPartition(p => {
    val poolPipelineMap = new ConcurrentHashMap[JedisPool, Pipeline](1024)
    val pipelineCount = new ConcurrentHashMap[Pipeline, Int](1024)
    val jedisList = new util.ArrayList[Jedis](1024)
    p.foreach(x => {
      // 自己封装的第一版JedisCluster的pipeline，后续又进行了优化
      val pipeline = clusterPipeline.value.initPipeline(x(0), poolPipelineMap, jedisList, pipelineCount, 512)
      if ("hmset".equals(operator)) {
        pipeline.hmset(x(0), clusterPipeline.value.file2Map(x))
      } else if ("set".equals(operator)) {
        pipeline.set(x(0), x(1))
      }
    })
    clusterPipeline.value.pipelineSync(pipelineCount)
    clusterPipeline.value.releaseConnection(jedisList)
  })

val conf = new SparkConf()
val duration = args(3).toInt
val ssc = new StreamingContext(conf, Seconds(duration))
ssc.checkpoint("./tmp")

val streams: DStream[String] = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    ).map(_.value)
    
val result = streams.map(x => x.split("\t"))
      .filter(x => x(0).startsWith("1"))
      // 生成(k,v)元组
      .map(x => (x(0), Integer.parseInt(x(21)) + Integer.parseInt(x(22)) + Integer.parseInt(x(24))
        + Integer.parseInt(x(25)) + Integer.parseInt(x(30)) + Integer.parseInt(x(31))))
      .reduceByKey((x, y) => x + y)
       // 状态计算
      .mapWithState(StateSpec.function((k: String, v: Option[Int], state: State[Int]) => {
        val state_flow = state.getOption().getOrElse(0)
        var flag = 0
        var sum_flow = 0
        if (!state.isTimingOut()) {
          val flow = v.get
          sum_flow = state_flow + flow
          val state_flow_M = state_flow / (1024 * 1024)
          val sum_flow_M = sum_flow / (1024 * 1024)
          if (state_flow_M < 1024 && sum_flow_M >= 1024) {
            flag = 3
          } else if (state_flow_M < 500 && sum_flow_M >= 500) {
            flag = 2
          } else if (state_flow_M < 200 && sum_flow_M >= 200) {
            flag = 1
          }
          state.update(sum_flow)
        }
        Option((k, flag))
      }).timeout(Minutes(182))
      )

val result = p.toList.map(x => {
  val userID = x._1
  val flow = x._2
  val response_date = pipeline.hget(userID, "date")
  // 从redis中获取当前user的流量
  val response_flow_codis = pipeline.hget(userID, "flow")
  i += 2
  if (i % 512 ==0){
    pipeline.sync
  }
  (userID, flow, response_date, response_flow_codis)
})
pipeline.sync
i = 0
result.foreach(x => {
  val userID = x._1
  // 当前数据中的流量数据
  var flow = x._2
  val date = x._3.get
  // redis中历史的流量数据
  val flow_codis = x._4.get
  // 说明flow不是今天的，直接覆盖
  if (date == null || date != start_date.value) {
    pipeline.hset(userID, "flow", String.valueOf(flow))
    pipeline.hset(userID, "date", start_date.value)
  } else if (flow_codis != "NN" && date == start_date.value) {
    // 累加当前数据流量和redis历史流量，然后更新到redis
    flow = Integer.parseInt(flow_codis) + flow
    pipeline.hset("userID, "flow", String.valueOf(flow))
  }
})