寻求Scala2.11下Kafka Streaming RDD带zscore追加存Redis的方案

针对你遇到的Spark-Redis连接器版本兼容、文档缺失的问题，这里有几个实用的方案，直接贴合你的现有代码框架：

方案1：直接使用Jedis原生客户端（最灵活可控） #

跳过第三方Spark连接器，直接在RDD的foreachPartition中用Jedis操作Redis有序集合（ZSet），这种方式不受Scala版本限制，完全自定义逻辑，还能避免依赖兼容问题。

步骤： #

1. 引入Jedis依赖（Maven为例）：

<dependency>
    <groupId>redis.clients</groupId>
    <artifactId>jedis</artifactId>
    <version>4.4.3</version> <!-- 选最新稳定版即可 -->
</dependency>

2. 修改你的现有代码：
   利用foreachPartition减少Redis连接创建次数（比foreach高效很多），每个分区内创建一次连接，批量执行ZADD命令（这就是Redis有序集合的追加模式，自动关联ZScore）：

import redis.clients.jedis.Jedis

collection.foreachRDD(rdd => {
  if (!rdd.partitions.isEmpty) {
    rdd.foreachPartition(partition => {
      // 每个分区创建一次Redis连接，避免频繁建连
      val jedis = new Jedis("your-redis-host", 6379)
      // 有密码的话加上认证
      // jedis.auth("your-redis-password")
      
      // 遍历分区内的每个元素，假设你的RDD结构是(key, (value, zscore))
      partition.foreach { case (redisKey, (memberValue, zscore)) =>
        // 往指定key的有序集合追加元素，memberValue是集合成员，zscore是对应的分数
        jedis.zadd(redisKey, zscore, memberValue)
      }
      
      // 用完关闭连接
      jedis.close()
    })
  }
})

注：如果你的RDD元素结构不是(key, (value, zscore))，可以根据实际数据格式调整foreach里的逻辑，核心就是调用Redis的ZADD命令。

方案2：使用Apache Bahir的Spark-Redis连接器（官方维护，兼容性好） #

Apache Bahir提供了官方维护的Spark与Redis集成组件，支持Scala 2.11/2.12/2.13，文档完善，Maven可直接拉取依赖，完全解决你之前遇到的“无Jar包、无文档”问题。

步骤： #

1. 引入依赖（Maven为例，对应你的Spark版本选合适的版本号）：

<dependency>
    <groupId>org.apache.bahir</groupId>
    <artifactId>spark-redis_2.12</artifactId>
    <version>3.0.0</version> <!-- Spark 3.x用这个，Spark 2.x用2.4.0 -->
</dependency>

2. 配置Redis连接参数：
   在SparkConf中添加Redis的连接信息：

val conf = new SparkConf()
  .setAppName("KafkaStreamToRedis")
  .setMaster("local[*]") // 生产环境替换为集群地址
  .set("redis.host", "your-redis-host")
  .set("redis.port", "6379")
  // 有密码的话加上认证配置
  // .set("redis.auth", "your-redis-password")

3. 修改代码写入Redis有序集合：
   将RDD转换为RedisZSet类型的RDD，然后用saveAsRedisDataset写入（ZSet本身就是追加模式，重复成员会自动更新分数）：

import org.apache.spark.streaming.redis._
import org.apache.spark.streaming.redis.api.java.RedisZSet

collection.foreachRDD(rdd => {
  if (!rdd.partitions.isEmpty) {
    // 把你的RDD转换为RedisZSet要求的格式
    val zsetRDD = rdd.map { case (redisKey, (memberValue, zscore)) =>
      RedisZSet(redisKey, memberValue, zscore)
    }
    // 直接写入Redis，自动追加到有序集合
    zsetRDD.saveAsRedisDataset()
  }
})

注：如果你的数据流是DStream，也可以直接调用dstream.saveAsRedisZSet("fixed-key")，前提是DStream的元素格式是(memberValue, zscore)。

方案3：自定义Spark-Redis写入工具类（适合特殊业务需求） #

如果以上方案都不满足你的定制化要求，你可以基于Jedis封装一个可复用的写入工具类，方便在多个地方调用：

object RedisZSetWriter {
  def write(rdd: RDD[(String, (String, Double))], redisHost: String, redisPort: Int, auth: Option[String] = None): Unit = {
    rdd.foreachPartition(partition => {
      val jedis = new Jedis(redisHost, redisPort)
      auth.foreach(jedis.auth)
      partition.foreach { case (key, (value, score)) =>
        jedis.zadd(key, score, value)
      }
      jedis.close()
    })
  }
}

// 在你的代码中直接调用
collection.foreachRDD(rdd => {
  if (!rdd.partitions.isEmpty) {
    RedisZSetWriter.write(rdd, "your-redis-host", 6379, Some("your-password"))
  }
})

额外生产环境提示： #

• 建议使用Jedis连接池替代单连接，避免频繁创建销毁连接带来的性能损耗；
• 如果ZScore需要实时计算，可以在RDD转换阶段先完成ZScore的计算，再传入写入逻辑；
• Redis ZSet的重复成员会自动更新分数，如果你需要保留历史数据，可以给成员加上时间戳后缀来区分。

内容的提问来源于stack exchange，提问作者Pinnacle