SparkStreaming——在RDD中查询redis
问题描述:在读取kafka数据时需要从redis查询出来上一条数据和当前数据进行计算。解决步骤:1.进入依赖<!-- https://mvnrepository.com/artifact/com.redislabs/spark-redis --><dependency><groupId>com.redislabs</groupId><artif
·
问题描述:
在读取kafka数据时需要从redis查询出来上一条数据和当前数据进行计算。
解决步骤:
1.进入依赖
<!-- https://mvnrepository.com/artifact/com.redislabs/spark-redis -->
<dependency>
<groupId>com.redislabs</groupId>
<artifactId>spark-redis</artifactId>
<version>2.4.0</version>
</dependency>
2.设置sparkConf
//设置参数
SparkConf conf = new SparkConf();
conf.set("spark.redis.host", "192.168.144.153"); //redis 主机节点
conf.set("spark.redis.port", "6379"); //端口号,不填默认为6379
// conf.set("spark.redis.auth","null"); //用户权限配置
conf.set("spark.redis.db","2"); //数据库设置
3.拿到redis连接,查询数据
Jedis jedis = ConnectionPool.connect(new RedisEndpoint(conf));
//查询redis中的数据
String realDataLastStr = jedis.hget(RedisKey.HASH_VEHICLE_REAL_DATA, vin);
4.查出来的数据是String需要转换成对象,JSONUtil工具类使用的hutool的
RealtimeDataHB realtimeDataHBLast = JSONUtil.toBean(realDataLastStr, RealtimeDataHB.class);
搞定了!
运行过程中可能会报错
[ WARN ] 2022-01-05 13:48:21 [ driver-heartbeater:23640 ] [org.apache.spark.internal.Logging$class.logWarning(Logging.scala:69)] Exception when trying to compute pagesize, as a result reporting of ProcessTree metrics is stopped
出现这个问题需要在适当的位置关闭掉数据库连接就可以了,或者直接将连接定义在try中
更多推荐


所有评论(0)