Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add expire to spark-redis #55

Open
wants to merge 2 commits into
base: master
Choose a base branch
from
Open
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Next Next commit
Add expire to spark-redis
  • Loading branch information
hanibash committed Jun 8, 2017
commit 6bc67bdf69b08ac3e7a5d13a4c639943fa984069
24 changes: 24 additions & 0 deletions src/main/scala/com/redislabs/provider/redis/redisFunctions.scala
Original file line number Diff line number Diff line change
Expand Up @@ -217,6 +217,14 @@ class RedisContext(@transient val sc: SparkContext) extends Serializable {
}
}

/**
* @param kttls Pair RDD of K/TTL
*/
def toRedisEXPIRE(kttls: RDD[(String, Int)])
(implicit redisConfig: RedisConfig = new RedisConfig(new RedisEndpoint(sc.getConf))) {
kttls.foreachPartition(partition => setExpire(partition, redisConfig))
}

/**
* @param kvs Pair RDD of K/V
* @param ttl time to live
Expand Down Expand Up @@ -283,6 +291,22 @@ class RedisContext(@transient val sc: SparkContext) extends Serializable {


object RedisContext extends Serializable {
/**
* @param arr k/ttl which should be set to expire in the target host
*/
def setExpire(arr: Iterator[(String, Int)], redisConfig: RedisConfig) {
arr.map(kv => (redisConfig.getHost(kv._1), kv)).toArray.groupBy(_._1).
mapValues(a => a.map(p => p._2)).foreach {
x => {
val conn = x._1.endpoint.connect()
val pipeline = conn.pipelined
x._2.foreach(x => pipeline.expire(x._1, x._2))
pipeline.sync
conn.close
}
}
}

/**
* @param arr k/vs which should be saved in the target host
* save all the k/vs to the target host
Expand Down