背景描述

新用户是一个很重要的指标，在数据量、用户量庞大的情况下，判断一个用户是否新用户，也需要耗费不少资源。

用户信息（5.07亿）存在 tidb 中，在数据清洗（kafka2kafka）时，查询 tidb 判断是否新用户，每日数据量 20-30 亿，就会有 20-30亿的请求，导致 tidb 负载高

调研

结合实际场景，调研了 redis-bitmap、redis-bloom、redis-hyperloglog 这三款高效数据存储

redis-bitmap

bitmap的底层数据结构就是0/1bit的二进制数值，最大长度512Mb = 512 * 1024 * 1024 * 8 = 2^32，但是支持数值类型

如果您的用户量几千万、几亿、甚至几十亿，通过redis-bitmap，不超过512Mb存储空间就可以做到了，不仅高效，且准确性100%，支持42亿+用户

如何有1亿用户量，每个用户30个字节来计算（redis哈希还需要耗费资源），用set存储，耗费资源：3000000000/8/1024/1024=357MB，如果用户量达到5亿，就得耗费1.74GB来存储，而且效率远远不如bitmap

# 创建一个 tmp_key bitmap，并将 1 的位置设置为1
127.0.0.1:6379> SETBIT tmp_key 1 1
(integer) 0
# 判断是否存在于bitmap，1存在返回1，2不存在返回0
127.0.0.1:6379> GETBIT tmp_key 1
(integer) 1
127.0.0.1:6379> GETBIT tmp_key 2
(integer) 0
127.0.0.1:6379> MEMORY USAGE tmp_key
(integer) 52

# 存储一个数值比较大的 ，redis底层会自动扩容，所以耗费时间比较久1.52s，再继续插入就不会了
127.0.0.1:6379> SETBIT tmp_key 4200000000 1
(integer) 0
(1.52s)
127.0.0.1:6379> SETBIT tmp_key 4200000001 1
(integer) 0
# key资源占用就上来了
127.0.0.1:6379> MEMORY USAGE tmp_key
(integer) 526048636

# 才存几个用户，就占用那么多磁盘空间，这也太离谱了吧，所以得用在合适场景
# 用户 1、2......4200000000,中间的位都是0，也是要消耗存储的；升级版大哥 roaring bitmap，在数据比较稀疏（数据量比较少，数值本身又比较大）场景时，
# 用计算资源换存储空间，roaring bitmap使用分桶存储（32位存储 -> 2个16位存储），但是redis目前不支持 roaring bitmap 这种存储
# redis-roaring开源项目，但是作者只进行了微基准测试，也仅只支持32位

如果您的用户标识是uid（数值类型），使用bitmap非常合适；不巧的是，公司基于 vid（一串字符，类似于uuid）标识用户，所以就尝试基于vid去做hash，hash必然有冲突，准确性就夭折了，以下是测试结果：

Hash算法	Hash数据量	冲突率
bkdrhash	5.07亿	5.46%
sdbmhash		5.46%
djbhash		5.46%
murmurHash		5.45%
fnvhash		5.45%

这个冲突率太高了，所以此方案是行不通的，从而引申出 redis bloom 插件，如下介绍

redis-bloom

概率型数据结构

Redis Bloom Filter 底层也是基于位数组 + N个哈希函数实现的，并采用了一些优化策略，使得其可以支持更大的数据规模和更低的误判率

当查询一个元素时，通过对该元素进行相同的哈希计算，检查对应的位是否都为1。如果其中有任何一位为0，则可以确定该元素不在集合中；如果所有位都为1，则该元素可能在集合中，但并不确定，存在一定的概率误判

优缺点：内存占用小、效率高，但是存在一定误差、无法删除

适用场景：Bloom 过滤器适用于需要快速判断元素是否属于一个大规模集合的场景

适用范围：Bloom 过滤器适用于对查询时间和空间占用有较高要求，而对概率误判可以接受的场景。在对精确性要求较高的情况下，Bloom 过滤器可能不适用

内存占用：

1%的错误率要求每项10.08位

0.1%的错误率要求每项14.4位

0.01%的错误率要求每项20.16位

2	预计插入数据量10亿，实际插入5.04亿	设置误判率0.01，实际误判率：5082/504858451=0.001006%	1.28GB
3	预计插入数据量100w，实际插入999325	设置误判率0.01，实际误判率：675/100w=0.0675%	1.3MB
4	预计插入数据量1000w，实际插入9992622	设置误判率0.01，实际误判率：7378/1000w=0.07378%	13.15MB

命令：

BF.RESERVE gamebox_bf 0.01 1000000000	创建gamebox_bf布隆过滤器，设置误判率为0.01，预计插入数据量10亿
bf.add gamebox_bf vid	插入数据，插入成功返回true，插入失败（已经存在）返回0

redis-hyperloglog

hyperloglog 基数统计概率算法

一种基数统计的算法，他的优势在于可以在输入元素的数量或者体积非常非常大时，计算基数所需的空间总是固定的、并且是很小的

在Redis里面，每个HyperLogLog键只需要花费12KB内存，就可以计算接近2^64个不同元素的基数。这和计算基数时，元素越多耗费内存就越多的集合形成鲜明对比，但是统计存在一定的误差，误差率整体较低，标准误差为 0.81%；

优缺点：节省空间，基于内存算法计算，相应查询插入响应没那么快，不如bloom，只能统计基数大小，不能是否存在redis-key中

使用场景：统计网站访问数、在线用户数

写入测试：

插入数据量	误判率	内存
插入100w 实际统计 998705	1295/100w=0.1295%	12.08KB
插入1000w 实际统计 9962589	37411/1kw=0.3741%	12.08KB

总结：虽然节省内存，但是hyperloglog只适用于统计基数大小，而不能用于判断一个元素存在于redis-key中

个人理解误区：

命令 pfadd key value，返回值：1表示 hyperloglog内部存储被修改了、0表示 hyperloglog内部存储未被修改

不能理解为是否新增，hyperloglog只是个概率算法，而非精确查找

使用组件版本：hadoop-3.3.0、spark-3.0.1、redis5.0.8、RedisBloom-2.2.6

# 启动spark-shell
bin/spark-shell --master yarn --packages org.springframework.data:spring-data-redis:2.3.9.RELEASE,org.apache.commons:commons-pool2:2.11.1,io.lettuce:lettuce-core:5.1.8.RELEASE

import java.util.Collections
import org.apache.spark.sql.{Row, SparkSession}
import org.apache.commons.pool2.impl.GenericObjectPoolConfig
import org.springframework.data.redis.connection.RedisStandaloneConfiguration
import org.springframework.data.redis.connection.lettuce.{LettuceConnectionFactory, LettucePoolingClientConfiguration}
import org.springframework.data.redis.core.StringRedisTemplate
import org.springframework.data.redis.core.script.{DefaultRedisScript}
import scala.collection.mutable.ListBuffer

def batchAddInBloomLua = {
  val sb = new StringBuilder
  sb.append("for index, arg in pairs(ARGV)").append("\r\n")
  sb.append("do").append("\r\n")
  sb.append("redis.call('bf.add', KEYS[1], arg)").append("\r\n")
  sb.append("end").append("\r\n")
  sb.append("return true")
  sb.toString
}

def batchAddInBloomFilter(stringRedisTemplate: StringRedisTemplate, key: String, valueList: ListBuffer[String]): Boolean = {
  val script = new DefaultRedisScript[Boolean](batchAddInBloomLua, classOf[Boolean])
  stringRedisTemplate.execute(script, Collections.singletonList(key),valueList: _*)
}

def existInBloomLua = "return redis.call('bf.exists', KEYS[1], ARGV[1])"

def existInBloomFilter(stringRedisTemplate: StringRedisTemplate, key: String, arg: Object): Boolean = {
  val script = new DefaultRedisScript(existInBloomLua, classOf[Boolean])
  stringRedisTemplate.execute(script, Collections.singletonList(key), arg)
}

val _count1 = spark.sparkContext.longAccumulator("_count1")
val _count2 = spark.sparkContext.longAccumulator("_count2")

def save2redis() = (rows: Iterator[Row]) => {
  val redisStandaloneConfiguration = new RedisStandaloneConfiguration("ip", 6399)
  val genericObjectPoolConfig = new GenericObjectPoolConfig
  genericObjectPoolConfig.setMaxIdle(8)
  genericObjectPoolConfig.setMinIdle(0)
  genericObjectPoolConfig.setMaxTotal(8)
  val builder = LettucePoolingClientConfiguration.builder
  builder.poolConfig(genericObjectPoolConfig)
  val connectionFactory = new LettuceConnectionFactory(redisStandaloneConfiguration, builder.build())
  connectionFactory.afterPropertiesSet()
  val stringRedisTemplate = new StringRedisTemplate
  stringRedisTemplate.setConnectionFactory(connectionFactory)
  stringRedisTemplate.afterPropertiesSet()
  val tmpValueArray = ListBuffer[String]()
  try {
    rows.foreach(row => {
      if (existInBloomFilter(stringRedisTemplate, "redis-key", row.getString(0))) {
        _count1.add(1)
      } else {
        _count2.add(1)
      }
    })
    if (tmpValueArray.size > 0) {
      batchAddInBloomFilter(stringRedisTemplate, "redis-key", tmpValueArray)
      tmpValueArray.clear()
    }
  } catch {
    case e: Throwable => {
      e.printStackTrace()
    }
  }
  finally {

  }
}



val vid_old = spark.sql(
  s"""
     |select vid from database.table where datekey='20240606' and first_datekey='20240602' and vid not REGEXP '[^a-zA-Z0-9 ]'
     |""".stripMargin).cache()
     
vid_old.coalesce(1).foreachPartition(save2redis())

println("老用户：" + _count1)
println("新用户：" + _count2)