使用spark-redis组件访问云数据库Redis

Posted 2021-04-29 Apache Spark技术交流社区

---

本文演示了在Spark Shell中通过spark-redis组件读写Redis数据的场景。所有场景在阿里云E-MapReduce集群内完成，Redis使用阿里云数据库Redis。

创建服务

我们以EMR-3.21.0版本和Redis 4.0为例。EMR集群安装的Spark版本是2.4.3，我们需要使用对应的Spark-Redis 2.4版本，该组件可以支持Redis 2.9.0以上版本。

启动Spark Shell

接下去，我们登录EMR Master节点启动Spark Shell。如果Master节点可以连接外网，可以使用package方式加载spark-redis相关jar包：

spark-shell --packages com.redislabs:spark-redis:2.4.0 \ --conf spark.redis.host=hostname \ --conf spark.redis.port=6379 \ --conf spark.redis.auth=password

spark.redis.host等参数可以在命令行指定，也可以配置在 spark-defaults.conf 中，也可以在代码中指定。其中：

2. spark.redis.port：Redis服务端口号

3. spark.redis.auth：创建Redis实例时指定的密码

也可以通过--jars的方式指定依赖的jar包：

spark-shell --jars spark-redis-2.4.0.jar,jedis-3.1.0-m1.jar,commons-pool2-2.0.jar \ --conf spark.redis.host=hostname \ --conf spark.redis.port=6379 \ --conf spark.redis.auth=password

通过Spark写入数据到Redis（RDD）

scala> import com.redislabs.provider.redis._ import com.redislabs.provider.redis._
scala> val data = Array(("key1", "v1"), ("key2", "world"), ("key3", "hello"), ("key4", "Hong"), ("key5", "Kong")) data: Array[(String, String)] = Array((key1,v1), (key2,world), (key3,hello), (key4,Hong), (key5,Kong))
scala> val distData = sc.parallelize(data) distData: org.apache.spark.rdd.RDD[(String, String)] = ParallelCollectionRDD[0] at parallelize at <console>:29
scala> sc.toRedisKV(distData)

读取Redis（RDD）

scala> val stringRDD = sc.fromRedisKV("key*").map{ kv => kv._2 }stringRDD: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[3] at map at <console>:27
scala> val values = stringRDD.collect()values: Array[String] = Array(world, hello, v1, Kong, Hong)
scala> println(values.mkString(","))world,hello,v1,Kong,Hong

Spark DataFrame写入Redis

scala> case class Person(name: String, age: Int)defined class Person
scala> val personSeq = Seq(Person("John", 30), Person("Peter", 45))personSeq: Seq[Person] = List(Person(John,30), Person(Peter,45))
scala> val df = spark.createDataFrame(personSeq)df: org.apache.spark.sql.DataFrame = [name: string, age: int]
scala> df.write.format("org.apache.spark.sql.redis").option("table", "person").save() 

参考文档

更多使用spark-redis的方式请参考官方文档：

1. spark-redis Package：

   https://spark-packages.org/package/RedisLabs/spark-redis

2. spark-redis Github：

   https://github.com/RedisLabs/spark-redis

   
   

---

对开源大数据感兴趣的同学可以加小编微信（图一二维码，备注进群）进入技术交流微信2群。也可钉钉扫码加入社区的钉钉群

阿里巴巴开源大数据技术团队成立Apache Spark中国技术社区，定期推送精彩案例，技术专家直播，问答区数个Spark技术同学每日在线答疑，只为营造纯粹的Spark氛围，欢迎钉钉扫码加入！