本篇内容主要讲解“如何使用spark-redis组件访问云数据库Redis”,感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷,实用性强。下面就让小编来带大家学习“如何使用spark-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快速入门文档,https://help.aliyun.com/document_detail/107043.html)。
接下去,我们登录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 中,也可以在代码中指定。其中:
spark.redis.host:Redis内网连接地址
spark.redis.port:Redis服务端口号
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
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)
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
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组件访问云数据库Redis”有了更深的了解,不妨来实际操作一番吧!这里是亿速云网站,更多相关内容可以进入相关频道进行查询,关注我们,继续学习!
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。