SparkStreaming的两种wordCount

素颜马尾好姑娘i 2023-05-28 08:23 24阅读 0赞

object HelloWord01 { 
    
      def main(args: Array[String]): Unit = { 
        //基本配置
        val conf = new SparkConf().setMaster("local[*]").setAppName("HelloWord01")
        val ssc = new StreamingContext(conf, Seconds(3))
        ssc.sparkContext.setLogLevel("WARN")
        //从kafka读取数据
        val zk = "127.0.0.1:2181"
        val group = "sunpls"
        val topic = "wuhanPolice"
        val numThread = 2
        val topicMap = topic.split(",").map((_, numThread)).toMap
        val dataDStream = KafkaUtils.createStream(ssc, zk, group, topicMap, StorageLevel.MEMORY_AND_DISK_SER)
        //(word,1)
        val word_1 = dataDStream.flatMap { 
          case (key, value) => { 
            val arr = value.split(" ")
            arr
          }
        }.map { 
          case (value) => { 
            (value, 1)
          }
        }
        word_1.reduceByKey(_+_).print()
        ssc.start()
        ssc.awaitTermination()
    
      }
    
    }

有状态的，会叠加每个分区

object HelloWord02 { 
    
      def main(args: Array[String]): Unit = { 
        //基本配置
        val conf = new SparkConf().setMaster("local[*]").setAppName("HelloWord02")
        val ssc = new StreamingContext(conf, Seconds(3))
        ssc.sparkContext.setLogLevel("WARN")
        //由于会保存状态，一般保存到磁盘文件
        ssc.checkpoint("cp")
        //从kafka读取数据
        val zk = "127.0.0.1:2181"
        val group = "sunpls2"
        val topic = "wuhanPolice"
        val numThread = 2
        val topicMap = topic.split(",").map((_, numThread)).toMap
        val dataDStream = KafkaUtils.createStream(ssc, zk, group, topicMap, StorageLevel.MEMORY_AND_DISK_SER)
        val word_1: DStream[(String, Int)] = dataDStream.flatMap { 
          case (key, value) => { 
            val arr = value.split(" ")
            arr
          }
        }.map((_,1))
        //叠加从头到尾的结果
        word_1.updateStateByKey((seq:Seq[Int],buffer:Option[Int]) => { 
          val sum = buffer.getOrElse(0) + seq.sum
          Option(sum)
        }).print()
        ssc.start()
        ssc.awaitTermination()
    
      }
    }