sparkstreaming聚合类操作写入redis数据库

Spark Streaming可以通过DStream的reduceByKey、updateStateByKey、window等聚合类操作对数据进行聚合，然后将结果写入Redis数据库。

具体步骤如下：

1. 创建Redis连接池

import redis.clients.jedis.{JedisPool, JedisPoolConfig}

val redisHost = "localhost"
val redisPort = 6379
val redisTimeout = 30000
val jedisPoolConfig = new JedisPoolConfig()
val jedisPool = new JedisPool(jedisPoolConfig, redisHost, redisPort, redisTimeout)

2. 定义写入Redis的函数

def writeToRedis(iterator: Iterator[(String, Int)]): Unit = {
  val jedis = jedisPool.getResource
  iterator.foreach { case (key, value) =>
    jedis.hincrBy("wordCount", key, value)
  }
  jedis.close()
}

3. 使用reduceByKey等聚合类操作对数据进行聚合

val lines = ssc.socketTextStream("localhost", 9999)
val words = lines.flatMap(_.split(" "))
val pairs = words.map(word => (word, 1))
val wordCounts = pairs.reduceByKey(_ + _)

4. 将聚合结果写入Redis

wordCounts.foreachRDD(rdd => {
  rdd.foreachPartition(writeToRedis)
})

5. 关闭连接池

ssc.start()
ssc.awaitTermination()
jedisPool.destroy()

这样就可以将Spark Streaming聚合类操作的结果写入Redis数据库了。