spark 连接redis
时间: 2023-08-29 09:13:45 浏览: 153
在 Spark 中连接 Redis 可以使用 Redis 的 Java 客户端 Jedis 或者 Redisson。以下是使用 Jedis 连接 Redis 的示例代码:
1. 导入 Jedis 依赖
```xml
<dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
<version>3.6.0</version>
</dependency>
```
2. 创建 SparkSession
```scala
import org.apache.spark.sql.SparkSession
val spark = SparkSession.builder()
.appName("RedisExample")
.master("local[*]")
.getOrCreate()
```
3. 创建 Jedis 连接池
```scala
import redis.clients.jedis.{JedisPool, JedisPoolConfig}
val jedisPoolConfig = new JedisPoolConfig()
jedisPoolConfig.setMaxTotal(10)
jedisPoolConfig.setMaxIdle(5)
val jedisPool = new JedisPool(jedisPoolConfig, "localhost", 6379)
```
4. 从 Redis 中读取数据
```scala
import redis.clients.jedis.Jedis
val jedis: Jedis = jedisPool.getResource
val value: String = jedis.get("key")
jedis.close()
```
5. 将 RDD 写入 Redis
```scala
import redis.clients.jedis.Jedis
val rdd = spark.sparkContext.parallelize(Seq(
("key1", "value1"),
("key2", "value2"),
("key3", "value3")
))
rdd.foreachPartition(partition => {
val jedis: Jedis = jedisPool.getResource
partition.foreach(item => jedis.set(item._1, item._2))
jedis.close()
})
```
注意:以上示例中的连接池是单例模式,实际上在生产环境中需要根据实际情况调整连接池的大小和配置,避免连接池过大或过小的问题。
阅读全文