Loosing key context when using 'fromRedisHash'

[yananielsen]

Hi,
I am trying to load redis data of the form (the hash keys don't have a fixed value):
hset id_1 feature_1 value_1 hset id_1 feature_2 value_2 hset id_2 feature_2 value_2 hset id_3 feature_3 value_3
to a dataframe with the schema (id, feature,value) but the method fromRedisHash("*") returns only the hash:

Is there a way to retrieve the context? assuming I don't want to use the dataframe approach because it will require more memory (i.e hset id_3 feature feature_3 value value_3)

[fe2s]

Hi @yananielsen ,

You are right, we lack RDD API to read hashes with keys. We will consider introducing it in the next versions.

As a workaround you can create a custom RDD:

import org.apache.spark.sql.SparkSession
import com.redislabs.provider.redis._
import com.redislabs.provider.redis.partitioner.RedisPartition
import com.redislabs.provider.redis.rdd.{Keys, RedisKVRDD, RedisKeysRDD}
import org.apache.spark.{Partition, TaskContext}
import org.apache.spark.rdd.RDD
import scala.collection.JavaConverters._


 class RedisHashWithKeyRdd(
                             prev: RDD[String],
                             implicit val readWriteConfig: ReadWriteConfig
                           ) extends RDD[(String,(String, String))](prev) with Keys {

    override def getPartitions: Array[Partition] = prev.partitions

    override def compute(split: Partition, context: TaskContext): Iterator[(String,(String, String))] = {
      val partition: RedisPartition = split.asInstanceOf[RedisPartition]
      val sPos = partition.slots._1
      val ePos = partition.slots._2
      val nodes = partition.redisConfig.getNodesBySlots(sPos, ePos)
      val keys = firstParent[String].iterator(split, context)
      getHASHWithKey(nodes, keys)
    }

    def getHASHWithKey(nodes: Array[RedisNode], keys: Iterator[String]): Iterator[(String, (String, String))] = {

      groupKeysByNode(nodes, keys).flatMap { case (node, nodeKeys) =>
        val conn = node.endpoint.connect()
        val hashKeys = filterKeysByType(conn, nodeKeys, "hash")
        val res = hashKeys.flatMap { k =>
          conn.hgetAll(k).asScala.map {
            case (f, v) => (k, (f, v))
          }
        }.iterator
        conn.close()
        res
      }
    }
  }

Usage:

val readWriteConfig = ReadWriteConfig.fromSparkConf(sc.getConf)
val keysRDD = sc.fromRedisKeyPattern("*")
val rdd = new RedisHashWithKeyRdd(keysRDD, readWriteConfig)
rdd.foreach(println)

Prints:

(id_3,(feature_3,value_3))
(id_1,(feature_1,value_1))
(id_2,(feature_2,value_2))

[yananielsen]

Thank you very much, just what I needed!
Weirdly, I had to unpack this case (f, v) => (k, (f, v)) into this case (f, v) => (k, f, v) because for some reason it was only returning one feature-value set per key..I probably mishandled an iterator or something but I needed it in the (k,f,v) form anyway so I didn't dig in..