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

本文涉及的产品
EMR Serverless StarRocks,5000CU*H 48000GB*H
EMR Serverless Spark 免费试用,1000 CU*H 有效期3个月
简介: 本文演示了在Spark Shell中通过spark-redis组件读写Redis数据的场景。所有场景在阿里云E-MapReduce集群内完成,Redis使用阿里云数据库Redis。

作者:无谓 阿里巴巴高级技术专家,2008年加入阿里巴巴集团,先后在B2B和阿里云工作,一直从事大数据和分布式计算相关研究,作为主要开发和运维人员经历了阿里内部大数据集群的上线和发展壮大,现在阿里云EMR团队,负责Spark、Hadoop等计算引擎研发。


本文演示了在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包:

 --conf spark.redis.host=hostname \
 --conf spark.redis.port=6379 \
 --conf spark.redis.auth=password
AI 代码解读

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

1.spark.redis.host:Redis内网连接地址
2.spark.redis.port:Redis服务端口号
3.spark.redis.auth:创建Redis实例时指定的密码

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

 --conf spark.redis.host=hostname \
 --conf spark.redis.port=6379 \
 --conf spark.redis.auth=password
AI 代码解读

通过Spark写入数据到Redis(RDD)

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)
AI 代码解读

读取Redis(RDD)

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
AI 代码解读

Spark DataFrame写入Redis

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()
                                                                                
AI 代码解读

参考文档

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

1.spark-redis Package:https://spark-packages.org/package/RedisLabs/spark-redis
2.spark-redis Github:https://github.com/RedisLabs/spark-redis

_

相关实践学习
基于EMR Serverless StarRocks一键玩转世界杯
基于StarRocks构建极速统一OLAP平台
快速掌握阿里云 E-MapReduce
E-MapReduce 是构建于阿里云 ECS 弹性虚拟机之上,利用开源大数据生态系统,包括 Hadoop、Spark、HBase,为用户提供集群、作业、数据等管理的一站式大数据处理分析服务。 本课程主要介绍阿里云 E-MapReduce 的使用方法。
目录
打赏
0
0
0
0
1336
分享
相关文章
|
2月前
|
Redis Pipeline介绍 ---- 提高操作Redis数据库的执行效率。
Redis Pipeline是提高Redis执行效率的重要技术,通过批量发送命令,显著减少了网络往返次数,提高了系统的吞吐量和性能。在实际应用中,合理使用Pipeline可以有效优化Redis的性能,特别是在需要批量操作的场景下。本文通过Python和Java的示例代码展示了如何实现和使用Redis Pipeline,为开发者提供了具体的操作指南。
68 16
优化微服务架构中的数据库访问:策略与最佳实践
在微服务架构中,数据库访问的效率直接影响到系统的性能和可扩展性。本文探讨了优化微服务架构中数据库访问的策略与最佳实践,包括数据分片、缓存策略、异步处理和服务间通信优化。通过具体的技术方案和实例分析,提供了一系列实用的建议,以帮助开发团队提升微服务系统的响应速度和稳定性。
大数据-104 Spark Streaming Kafka Offset Scala实现Redis管理Offset并更新
大数据-104 Spark Streaming Kafka Offset Scala实现Redis管理Offset并更新
94 0
云端问道21期方案教学-应对高并发,利用云数据库 Tair(兼容 Redis®*)缓存实现极速响应
云端问道21期方案教学-应对高并发,利用云数据库 Tair(兼容 Redis®*)缓存实现极速响应
使用访问指导(SQL Access Advisor)优化数据库业务负载
本文介绍了Oracle的SQL访问指导(SQL Access Advisor)的应用场景及其使用方法。访问指导通过分析给定的工作负载,提供索引、物化视图和分区等方面的优化建议,帮助DBA提升数据库性能。具体步骤包括创建访问指导任务、创建工作负载、连接工作负载至访问指导、设置任务参数、运行访问指导、查看和应用优化建议。访问指导不仅针对单条SQL语句,还能综合考虑多条SQL语句的优化效果,为DBA提供全面的决策支持。
126 11
【赵渝强老师】Spark生态圈组件
本文介绍了Spark的生态圈体系架构,包括其核心执行引擎Spark Core、结构化数据处理模块Spark SQL、实时数据流处理模块Spark Streaming,以及机器学习框架MLlib和图计算框架GraphX。文中通过图片和视频详细解析了各模块的功能及访问接口。
125 2
在Java应用中,数据库访问常成为性能瓶颈。连接池技术通过预建立并复用数据库连接,有效减少连接开销,提升访问效率
在Java应用中,数据库访问常成为性能瓶颈。连接池技术通过预建立并复用数据库连接,有效减少连接开销,提升访问效率。本文介绍了连接池的工作原理、优势及实现方法,并提供了HikariCP的示例代码。
111 3
打破瓶颈:利用Java连接池技术提升数据库访问效率
在Java应用中,数据库访问常成为性能瓶颈。连接池技术通过预建立并复用数据库连接,避免了频繁的连接建立和断开,显著提升了数据库访问效率。常见的连接池库包括HikariCP、C3P0和DBCP,它们提供了丰富的配置选项和强大的功能,帮助优化应用性能。
155 2
优化微服务架构中的数据库访问:策略与实践
随着微服务架构的普及,如何高效管理和优化数据库访问成为了关键挑战。本文探讨了在微服务环境中优化数据库访问的策略,包括数据库分片、缓存机制、异步处理等技术手段。通过深入分析实际案例和最佳实践,本文旨在为开发者提供实际可行的解决方案,以提升系统性能和可扩展性。
|
8月前
|
彻底革新你的数据库操作体验!Micronaut数据访问技巧让你瞬间爱上代码编写!
【9月更文挑战第10天】Java开发者们一直在寻找简化应用程序与数据库交互的方法。Micronaut作为一个现代框架,提供了多种工具和特性来提升数据访问效率。本文介绍如何使用Micronaut简化数据库操作,并提供具体示例代码。Micronaut支持JPA/Hibernate、SQL及NoSQL(如MongoDB),简化配置并无缝集成。通过定义带有`@Repository`注解的接口,可以实现Spring Data风格的命名查询。
135 6

热门文章

最新文章