scala - Spark : variable pipeline in class BinaryJedis cannot be accessed 上的 Redis

Question

我正在尝试从 spark 写入 redis。但是我收到一个编译时错误，提示“无法在 redis.clients.jedis.Jedis 中访问类 BinaryJedis 中的变量管道”。我的代码如下(部分展示):

    import org.sedis._
    import redis.clients.jedis._
    ...
    val myRDD = KafkaUtils.createStream(ssc, zkQuorum, group, topic).map(_._2).window(Seconds(300), Seconds(10))
    myRDD.foreachRDD( rdd => {rdd.foreachPartition(it =>{
      val pool = new Pool(new JedisPool(new JedisPoolConfig(), "localhost", 6379, 2000))
      pool.withJedisClient { client => 
        val pipeline = client.pipeline()
        it.foreach {
          case (a,b,c) => pipeline.hmset(a,Map("b" -> b, "c" -> c))
        }
      }

   })})

我得到的错误如下:

    error: variable pipeline in class BinaryJedis cannot be accessed in redis.clients.jedis.Jedis
    Access to protected variable pipeline not permitted because enclosing object MainExample in package examples is not a subclass of class BinaryJedis in package jedis where target is defined 
    val pipeline = client.pipeline()

我已经搜索了解决方案，但找不到。有人可以帮我吗？提前致谢。

Answer 1

上面的问题是通过删除分区并为每个写入redis的数据元组用一个新的jedis实例替换Jedis池来解决的，如下所示。这对我有用。

import org.sedis._
import redis.clients.jedis._
...
val myRDD = KafkaUtils.createStream(ssc, zkQuorum, group, topic)
                      .map(_._2)
                      .window(Seconds(300), Seconds(10))

myRDD.foreachRDD( rdd => {
      rdd.foreach( case(a, b, c) =>{
           val jedis = new Jedis("localhost", 6379)
           val pipeline = jedis.pipelined
           pipeline.hmset(a,Map("b" -> b, "c" -> c))
    })
})