redis基础spark操作redis

Redis内存淘汰策略

将Redis用作缓存时，如果内存空间用满，就会自动驱逐老的数据。

为什么要使用内存淘汰策略呢？

当海量数据涌入redis，导致redis装不下了咋办，我们需要根据redis的内存淘汰策略，淘汰一些不那么重要的key，来满足大量数据的存入。

Redis六种淘汰策略

noeviction：当内存使用达到阈值的时候，所有引起申请内存的命令会报错。
allkeys-lru：在主键空间中，优先移除最近未使用的key。(推荐)
volatile-lru：在设置了过期时间的键空间中，优先移除最近未使用的key。
allkeys-random：在主键空间中，随机移除某个key。
volatile-random：在设置了过期时间的键空间中，随机移除某个key。
volatile-ttl：在设置了过期时间的键空间中，具有更早过期时间的key优先移除。

如何配置Redis淘汰策略

找到redis.conf文件

设置Redis 内存大小的限制，我们可以设置maxmemory ，当数据达到限定大小后，会选择配置的策略淘汰数据比如：maxmemory 300mb。

设置内存淘汰具体使用那种策略

设置Redis的淘汰策略。比如：maxmemory-policy volatile-lru 或allkeys-lru

127.0.0.1:6379> CONFIG GET maxmemory-policy
1) "maxmemory"
2) "0"
127.0.0.1:6379> CONFIG SET maxmemory-policy allkeys-lru
OK
127.0.0.1:6379> CONFIG GET maxmemory-policy
1) "maxmemory-policy"
2) "allkeys-lru"

redis基本操作

查看连接数
CLIENT LIST获取客户端列表
CLIENT SETNAME 设置当前连接点redis的名称
CLIENT GETNAME 查看当前连接的名称
CLIENT KILL ip:port 杀死指定连接

查看资源占用情况

info查看全部
info memory查看资源占用情况

# Memory
used_memory:13490096 //数据占用了多少内存（字节）
used_memory_human:12.87M //数据占用了多少内存（带单位的，可读性好）
used_memory_rss:13490096  //redis占用了多少内存
used_memory_peak:15301192 //占用内存的峰值（字节）
used_memory_peak_human:14.59M //占用内存的峰值（带单位的，可读性好）
used_memory_lua:31744  //lua引擎所占用的内存大小（字节）
mem_fragmentation_ratio:1.00  //内存碎片率
mem_allocator:libc //redis内存分配器版本，在编译时指定的。有libc、jemalloc、tcmalloc这3种。
total_system_memory//整个系统内存
used_memory_dataset_perc //数据占用的内存大小的百分比，100%*(used_memory_dataset/(used_memory-used_memory_startup))
used_memory_peak_perc //使用内存达到峰值内存的百分比，即(used_memory/ used_memory_peak) *100%

--bigkeys可以查看下那些key比较占空间
redis-cli -c -h redis-recommend1.nonolive.local --bigkeys

键的数据结构类型

type key
如果键hello是字符串类型，则返回string；如果键不存在，则返回none

键重命名

rename key newkey
renamenx key newkey只有newkey不存在时才会被覆盖

查找key

不要使用keys xxx*查找，keys 算法是遍历算法，复杂度是 O(n)，如果实例中有千万级以上的 key，这个指令就会导致 Redis 服务卡顿，所有读写 Redis 的其它的指令都会被延后甚至会超时报错，可能会引起缓存雪崩甚至数据库宕机。
生产环境禁用
通常使用Scan模糊查找
scan 参数提供了三个参数，第一个是 cursor 整数值，第二个是 key 的正则模式，第三个是遍历的 limit hint。第一次遍历时，cursor 值为 0，然后将返回结果中第一个整数值作为下一次遍历的 cursor。一直遍历到返回的 cursor 值为 0 时结束。
SSCAN 命令用于迭代集合（set）键中的元素。
HSCAN 命令用于迭代哈希(hash)键中的键值对。
ZSCAN 命令用于迭代有序集合(sorted set)中的元素（包括元素成员和元素分值）

127.0.0.1:6379> scan 0 match key99* count 1000
1) "13976"        # 第一次迭代时返回的游标
2)  1) "key9911"
    2) "key9974"
    3) "key9994"
    4) "key9910"
    5) "key9907"
    6) "key9989" 

127.0.0.1:6379> scan 13976 match key99* count 1000 # 第二次查就使用第一次返回的游标

这里的count 1000不是限定的数量，而是限定服务器单次遍历的字典槽位数量 (约等于)，所以只要第一条有数，那就是有数据的。

然后可以通过key来核对数据

//type返回给定key的value类型
type key
返回 none 表示不存在key。string字符类型，list 链表类型 set 无序集合类型

//get返回给定key的value
get key

删除操作

删除前看一下是否存在：exists key
删除单个key: del key
删除所有key: flushdb

批量模糊删除：
如果需要批量删除，例如想一次性删除key为（redis_key1,redis_key2,redis_key3……）的这么一批redis_key，
退出redis客户端，直接在shell中使用xargs参数实现，
xargs 是一个强有力的命令，它能够捕获一个命令的输出，然后传递给另外一个命令。
如果需要指定数据库，需要用到 -n 数据库编号参数，下面是删除 2数据库中 redis_key开头的键：
redis-cli -n 数据库名(0-15) keys redis_key*|xargs redis-cli -n 数据库名(0-15) del
例子：
redis-cli -h 127.0.0.1 -p 6379 keys "redis_key*"|xargs redis-cli -h 127.0.0.1 -p 6379 del
补充的几个参数：
redis-cli keys "mailspec*" | xargs -r -t -n1 del
补充1：xargs命令后需加上参数-r,不然当keys的数量为0时，就会报错 (error) ERR wrong number of arguments for ‘del’ command
补充2：xargs命令后需加上参数-n1,不然当集群情况keys的数量大于1时，可能会报错 (error) CROSSSLOT Keys in request don’t hash to the same slot
补充3：不加-t也可以，加上-t会输出每次删除的内容，不加则不输出删除的内容，但还是会输出每次删除的key的数量
建议还是不要用keys，用scan模糊匹配更好些。

注意

del删除很大量的数据会把redis阻塞，导致其他命令无法执行。
和使用keys查找数据一样的结果。
生产环境禁用
建议使用unlink和scan异步来删，
redis-cli -h [ip] -p [port] -a [password] -n [index] --scan --pattern 'User:*' | xargs redis-cli -h [ip] -p [port] -a [password] -n [index] unlink
例如：
redis-cli -c -h redis-recommend1.nonolive.local --scan --pattern "REC_RECALL_OUTLINE_20210721*"|xargs -r -t -n1 redis-cli -c -h redis-recommend1.nonolive.local unlink
有时候数据模糊查找也太大，那就需要设置执行时间，用timeout删几分钟停一会儿再删几分钟。
使用timeout
timeout 300 redis-cli -c -h redis-recommend1.nonolive.local --scan --pattern "REC_RECALL_OUTLINE_20210721*"|xargs -r -t -n1 redis-cli -c -h redis-recommend1.nonolive.local unlink

unlink删除每删除一次会开一个redis连接，有些公司不允许开太多连接，那么也不能用unlink来删，只能重写一次数据把key的过期时间调为0。

有时候写入数据的时候报错：
WRONGTYPE Operation against a key holding the wrong kind of value
意思是写入的数据格式有问题，如果当前数据有问题就需要调整当前数据的格式，如果当前数据没问题，是因为redis中存在不同类型的数据，比如写入是hmset的hash类型，redis中存在相同的key的value格式为字符串，那么需要先删除这些key(批量删除或者直接删除)，再写入

redis过期命令(key)

Setex
指定的 key 设置值及其过期时间。如果 key 已经存在， SETEX 命令将会替换旧的值。
redis 127.0.0.1:6379> SETEX KEY_NAME TIMEOUT VALUE

Expire
设置 key 的过期时间，key 过期后将不再可用。单位以秒计。
redis 127.0.0.1:6379> Expire KEY_NAME TIME_IN_SECONDS

TTL
Redis TTL 命令以秒为单位返回 key 的剩余过期时间。key不存在返回-2，存在，返回-1，否则反回剩余生存时间。
redis 127.0.0.1:6379> TTL KEY_NAME

PERSIST
移除key的过期时间，key将持久保持：PERSIST key

hash数据结构常用命令

批量设置或获取field-value
Redis Hmset 命令用于同时将多个 field-value (字段-值)对设置到哈希表中。
此命令会覆盖哈希表中已存在的字段。
如果哈希表不存在，会创建一个空哈希表，并执行 HMSET 操作。
redis 127.0.0.1:6379> HMSET KEY_NAME FIELD1 VALUE1 ...FIELDN VALUEN
如果用python写入，value就是一个字典。

Redis Hmget 命令用于返回哈希表中，一个或多个给定字段的值。
如果指定的字段不存在于哈希表，那么返回一个 nil 值。
redis 127.0.0.1:6379> HMGET KEY_NAME FIELD1...FIELDN

设置值
Redis Hset 命令和Hmset类似，但是只能给一个field-value (字段-值)赋值。
如果哈希表不存在，一个新的哈希表被创建并进行 HSET 操作。
如果字段已经存在于哈希表中，旧值将被覆盖。
redis 127.0.0.1:6379> HSET KEY_NAME FIELD VALUE

获取所有的field、value
Redis Hgetall 命令用于返回哈希表中，所有的字段和值。
在返回值里，紧跟每个字段名(field name)之后是字段的值(value)，所以返回值的长度是哈希表大小的两倍。
redis 127.0.0.1:6379> HGETALL KEY_NAME

获取值hget key field
redis 127.0.0.1:6379> HGET KEY_NAME FIELD_NAME

判断field是否存在
hexists key field

字段值加上指定增量值
Redis Hincrby
如果哈希表的 key 不存在，一个新的哈希表被创建并执行 HINCRBY 命令。
如果指定的字段不存在，那么在执行命令前，字段的值被初始化为 0 。
对一个储存字符串值的字段执行 HINCRBY 命令将造成一个错误。
redis 127.0.0.1:6379> HINCRBY KEY_NAME FIELD_NAME INCR_BY_NUMBER

字段值加上指定小数增量值
Redis Hincrbyfloat
命令用于为哈希表中的字段值加上指定浮点数增量值。

spark连接redis
几种连接方法
单例模式：静态内部类

import org.apache.commons.pool2.impl.GenericObjectPoolConfig
import redis.clients.jedis.JedisPool
import redis.clients.jedis.JedisCluster
import redis.clients.jedis.HostAndPort

import java.util

/**
 * redis pool
 */
object RedisPool {

  val poolConfig = new GenericObjectPoolConfig()
  poolConfig.setMaxIdle(10)          // 最大连接数
  poolConfig.setMaxTotal(1000)       // 最大空闲连接数
  poolConfig.setMaxWaitMillis(1000)  // 最大等待时间
  poolConfig.setTestOnBorrow(true)   // 检查连接可用性, 确保获取的redis实例可用

  val host="redis-recommend.mildom.local"
  val PORT = 6379

  private lazy val jedisPool = new JedisPool(poolConfig, host, PORT)

  def getJedis() = {
    val jedis = jedisPool.getResource //获取连接池连接
    jedis
  }
}

分布式Jedispool

import redis.clients.jedis.JedisCluster
import redis.clients.jedis.HostAndPort

import java.util
/**
 * redis pool
 */
object RedisClusterPool {

  val host="redis-recommend.mildom.local"
  val PORT = 6379

  private val jedisClusterNodes = new util.HashSet[HostAndPort]()
  jedisClusterNodes.add(new HostAndPort(host, PORT))

  def getJedis() = {
    lazy val jedisCluster = new JedisCluster(jedisClusterNodes)
    jedisCluster
  }
}

spark写入redis
sparksql任务写入redis

//随机给一个过期时间用于测试
    val rng = new scala.util.Random
//df_result结构为（redis_key,redis_value）

    df_result.rdd.repartition(50).foreachPartition(partition=>{
      val jedis = RedisClusterPool.getJedis()
      partition.foreach(v=>{
        val redisKey=v(0).toString
        val redisValue=v(1).toString
        if (redisKey.length>0) {
          jedis.set(redisKey, redisValue)
          jedis.expire(redisKey, rng.nextInt(500))
        }
        else{
          println("no data")
        }
      })})

hmset 之类的hash格式在python中写入dict即可，scala中需要把scala的map转化成java的map，dataframe中可以保存map格式的数据，df转rdd时使用getAs写一下Map[...,...]类型，否则Map中的类型无法自动识别。
spark sql functions中提供了map方法，直接选取两列可以转为一个scala map。

  val df_result = sparkSession.sql(sql)
      .filter("region is not null and host_id is not null")
      .repartition(30)
      .withColumn("redis_key", lit(redis_key))
      .select(col("redis_key"), map(col("region"), col("host_id")).alias("redis_value"))
      .toDF("redis_key","redis_value")

    df_result.rdd.repartition(30).foreachPartition(partition=>{
      val jedis = nn_RedisClusterPool.getJedis()
      partition.foreach(v=>{
        val redisKey=v.getAs[String]("redis_key").toString
        import scala.collection.JavaConverters._
        // 同样将scala的map转换为Java的map存入redis中
        val map =  v.getAs[Map[String,String]]("redis_value")
        val redisValue_map: java.util.Map[String, String] = map.asJava
        if (redisKey.length>0) {
          try {
            jedis.hmset(redisKey, redisValue_map)
            jedis.expire(redisKey, seconds + rng.nextInt(5000))
          }
          catch {
            case e:Exception => e.printStackTrace()
          }
        }
        else{
          println("no data")
        }
      })})

spark streaming写入redis

//导入上面的redispool
import utils.stream.RedisClusterPool
val jedis = RedisClusterPool.getJedis()

stream.repartition(30).foreachRDD(rdd => {
val topic_expose =消费kafka曝光数据和其他离线数据组合

    if (topic_expose.count()>0){
      topic_expose.rdd.collect().foreach(v=>{
      jedis.hset("user:bought:data:"+v(0)+":"+v(1), "exposeCount", jedis.sadd("purchase_expose_num",v(0).toString+":"+v(2).toString).toString)
"exposeCount", 1)
     jedis.expire("user:bought:data:"+v(0)+":"+v(1),seconds)
        })
      }
    })

spark3.0亲测可用，但是topic_expose.rdd.collect()转化成数组写入redis需要先把数据拿到driver端，偶尔sparkstreaming的实时小数据任务可以，如果是离线或实时的大数据集，造成的shuffle和内存消耗是不可接受的。

spark在foreachRDD中初始化外部连接也有一些注意点，由于在foreachRDD中初始化了jedis，但是后续写入redis的时候也要用到foreach一条条数据往里面写，还是会有序列化的error，如果在foreach中初始化，非常浪费资源，也没有必要。
通常的做法是使用foreachPartition，在每个partition中初始化外部数据源。
正常写法：

import utils.stream.RedisClusterPool

stream.repartition(100).foreachRDD(rdd =>{
  rdd.foreachPartion(partition=>{
    val topic_expose =消费kafka曝光数据和其他离线数据组合(组合了partition)
    val jedis = RedisClusterPool.getJedis()
    topic_expose.rdd.foreach(x=>{
    //用redis中的set,以去重的方式写入redis
    jedis.hset("user:bought:data:"+v(0)+":"+v(1), "exposeCount", jedis.sadd("purchase_expose_num",v(0).toString+":"+v(2).toString).toString)
    "exposeCount", 1)
    //设置过期时间
    jedis.expire("user:bought:data:"+v(0)+":"+v(1),seconds)
    })
  })

最后编辑于：2024-11-14 21:10:57