使用spark-redis元件通路雲資料庫Redis

本文示範了在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以上版本。

EMR和Redis需要在同一個VPC網絡中建立，同時，在雲資料庫Redis執行個體啟動之後，需要在“白名單設定”中添加EMR叢集IP位址（參考Redis

快速入門文檔

）。

啟動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 中，也可以在代碼中指定。其中：

1. spark.redis.host：Redis内網連接配接位址
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