devbox
花生_TL007
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Redis 布隆过滤器存储查询千万级用户数据_布隆过滤器可以存储多少数据

作者：花生_TL007 | 2024-04-21 11:04:54

赞

踩

布隆过滤器可以存储多少数据

实时业务需要计算新增用户，从 Hive 加载用户表到内存存储，但是几千万数据放到内存占用资源，查询效率也低。使用 Redis 位图进行存储，可以有效减少内存占用，提高查询效率。这里使用 Spark 查询， Redis 位图存储。

这里是 Scala + Spark local模式 + Redis 单节点测试。

测试了几种 Hash 算法，单一 Hash 算法效果或者效率都不是很满意，在万级到几十万级时，出现了多个碰撞。经过几种算法多次Hash，结果仍然不理想，在千万数据时，仍会有大量碰撞。

测试了 Redissom，碰撞率接近的情况下，Redisson 效率会低很多（千万数据/小时级）。

测试了多个质数组合：（61），（7，61），（7，11，13，61），（5，7，11，13，61），（61，83，113，151，211），（61，83，113，151，211，379）...

最后选择 murmurhash，经过 5 次 Hash（37, 41, 61, 79, 83）， 使用 256m 存储 4kw 数据， 31 个碰撞，可以接受。

主类

  1. import java.util.Properties
  2.  
  3. import com.sm.bloom.Bloom
  4. import com.sm.constants.Constants
  5. import com.sm.utils.{DateUtils, RedisUtils}
  6. import org.apache.log4j.{Level, Logger}
  7. import org.apache.spark.sql.SparkSession
  8. import org.slf4j.LoggerFactory
  9. import redis.clients.jedis.Jedis
  10.  
  11. /**
  12.   * Spark查询用户表,导入Redis缓存
  13.   *
  14.   * create by LiuJinHe 2020/3/3
  15.   */
  16. object ImportDataToRedis {
  17.   private val warehouseLocation = Constants.WAREHOUSE
  18.   private val logger = LoggerFactory.getLogger(this.getClass)
  19.   private var prop: Properties = new Properties
  20.   private var jedis: Jedis = _
  21.   private var startDate: String = _
  22.   private var endDate: String = _
  23.   private var count: Long = 0L
  24.  
  25.   def main(args: Array[String]): Unit = {
  26.     Logger.getLogger("org.apache.hadoop").setLevel(Level.WARN)
  27.     Logger.getLogger("org.apache.spark").setLevel(Level.INFO)
  28.     Logger.getLogger("org.spark_project.jetty").setLevel(Level.WARN)
  29.     
  30.     // hash 质数
  31.     val seeds = Array(37, 41, 61, 79, 83)
  32.  
  33.     if (args.length == 2) {
  34.       startDate = args(0)
  35.       endDate = args(1)
  36.     }
  37.  
  38.     val spark = initSparkSession
  39.  
  40.     logger.info(s"========== 全量导入新增用户表到 Redis ==========")
  41.  
  42.     val start = System.currentTimeMillis()
  43.  
  44.     // 加载 redis 配置,初始化 jedis
  45.     val redisProp = loadProp(Constants.REDIS_PROP)
  46.     val host = redisProp.getProperty("redis.host")
  47.     val port = redisProp.getProperty("redis.port").toInt
  48.     val timeout = redisProp.getProperty("redis.timeout").toInt
  49.     jedis = RedisUtils.init(host, port, timeout)
  50.  
  51.     // 用户位图
  52.     // 位图存储用户记录，key：用户表名， value：bitmap (offset,即 hash(core_account + game_id + package_id))
  53.     val userBit = redisProp.getProperty("redis.user.bit")
  54.  
  55.     // 初始化布隆过滤器，128m
  56.     lazy val bloom = new Bloom(1 << 31)
  57.  
  58.     // 临时参数
  59.     var data: Array[String] = null
  60.     var gamePackage: String = ""
  61.     var userId: String = ""
  62.     var offset: Array[Long] = null
  63.     var isExist: Boolean = true
  64.  
  65.     // 读取用户表数据
  66.     val sqlStr =
  67.       s"""
  68.          |select
  69.          | core_account, game_id, package_id, time
  70.          |from sm_data.dws_new_user_id
  71.          | where add_date >= '$startDate' and add_date < '$endDate'
  72.       """.stripMargin
  73.  
  74.     val dataFrame = spark.sql(sqlStr)
  75.     dataFrame.foreach(line => {
  76.       // 组合字段作为userId
  77.       userId = line.getInt(1) + "," + line.getInt(2) + "," + line.getString(0)
  78.  
  79.       // hash
  80.       offset = bloom.murmurHash3(userId, seeds)
  81.  
  82.       // 判断 redis 位图中是否存在
  83.       isExist = bloom.getbit(jedis, userBit, offset)
  84.       if (!isExist) {
  85.         // 如果不存在，设置位图对应位置1
  86.         bloom.setbit(jedis, userBit, offset)
  87.         count += 1
  88.       }
  89.     })
  90.  
  91.     spark.stop()
  92.     jedis.close()
  93.  
  94.     val end = System.currentTimeMillis()
  95.     logger.info(s"========== 新增总数: $count ==========")
  96.     logger.info(s"========== 耗时： ${(end - start) / 1000} 秒 ==========")
  97.  
  98.   }
  99.  
  100.   def initSparkSession: SparkSession = SparkSession.builder()
  101.     .appName(this.getClass.getSimpleName)
  102.     .master(Constants.SPARK_LOCAL_MODE)
  103.     .config("spark.sql.warehouse.dir", warehouseLocation)
  104.     .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
  105.     .config("spark.kryoserializer.buffer", "1024m")
  106.     .config("spark.kryoserializer.buffer.max", "2046m")
  107.     .config("spark.io.compression.codec", "snappy")
  108.     .enableHiveSupport()
  109.     .getOrCreate()
  110. }

 RedisUtils

  1. import org.apache.commons.pool2.impl.GenericObjectPoolConfig
  2. import redis.clients.jedis.{Jedis, JedisPool}
  3.  
  4. /**
  5.   * create by LiuJinHe 2020/3/3
  6.   */
  7. object RedisUtils extends Serializable {
  8.  
  9.   // @transient 类型修饰符，只能用来修饰字段。
  10.   // 在对象序列化过程中，被transient 标记的变量不会被序列化
  11.   @transient private var redisPool: JedisPool = _
  12.   var host: String = _
  13.   var port: Int = _
  14.   var timeout: Int = _
  15.  
  16.   def init(host: String, port: Int, timeout: Int): Jedis = {
  17.     redisPool = new JedisPool(new GenericObjectPoolConfig, host, port, timeout)
  18.     redisPool.getResource
  19.   }
  20. }

Bloom类

  1. import redis.clients.jedis.Jedis
  2. import scala.util.hashing.MurmurHash3
  3.  
  4.  
  5. /**
  6.   * create by LiuJinHe 2020/3/3
  7.   */
  8. class Bloom(size: Long) extends Serializable {
  9.   // 位图大小，默认128m
  10.   private val cap = if (size > 0) size else 1 << 31
  11.   private val seeds = Array(37, 41, 61, 79, 83)
  12.   private var indices:Array[Long] = _
  13.  
  14.   /**
  15.     * BKDRHash
  16.     * @param value 需要hash的字符串
  17.     * @param seed 随机数种子
  18.     * @return hash结果
  19.     */
  20.   def hash(str: String, seed: Int): Long = {
  21.     var result = 0L
  22.     val value = md5Hash(str)
  23.     for (i <- 0 until value.length) {
  24.       result = result * seed + value.charAt(i)
  25.     }
  26.     result = result & (cap - 1)
  27.     println("result： " + result)
  28.     result
  29.   }
  30.  
  31.    /**
  32.     * murmurHash
  33.     * @param value 需要hash的字符串
  34.     * @return hash后的结果数组
  35.     */
  36.   def murmurHash3(value: String): Long = {
  37.     var result = 0L
  38.     if (value != null && !value.isEmpty) {
  39.       for (i <- seeds.indices) {
  40.         result = Math.abs(MurmurHash3.stringHash(value, seeds(i))) % cap
  41.       }
  42.     }
  43.     result
  44.   }
  45.  
  46.   /**
  47.     * murmurHash
  48.     * @param seeds 传入murHash 中的 seed 的范围
  49.     * @return hash后的结果数组
  50.     */
  51.   def murmurHash3(value: String, seeds: Array[Int]): Array[Long] = {
  52.     indices = new Array[Long](seeds.length)
  53.     if (value != null && !value.isEmpty) {
  54.       for (i <- seeds.indices) {
  55.         indices(i) = Math.abs(MurmurHash3.stringHash(value, seeds(i))) % cap
  56.       }
  57.     }
  58.     indices
  59.   }
  60.  
  61.   def setbit(jedis: Jedis, key: String, indices: Array[Long]): Unit = {
  62.     for (index <- indices) {
  63.       jedis.setbit(key, index, true)
  64.     }
  65.   }
  66.  
  67.   def getbit(jedis: Jedis, key: String, indices: Array[Long]): Boolean ={
  68.     if (key == null && key.isEmpty ) return false
  69.     var isExists = true
  70.     for (index <- indices) {
  71.       isExists = isExists && jedis.getbit(key, index)
  72.     }
  73.     isExists
  74.   }
  75. }

pom.xml

  1.     <properties>
  2.         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  3.         <maven.compiler.source>1.8</maven.compiler.source>
  4.         <maven.compiler.target>1.8</maven.compiler.target>
  5.         <scala.version>2.11.12</scala.version>
  6.         <scala.binary.version>2.11</scala.binary.version>
  7.         <scala.complete.version>${scala.binary.version}.8</scala.complete.version>
  8.         <hadoop.version>2.7.6</hadoop.version>
  9.         <spark.version>2.4.0</spark.version>
  10.         <slf4j.version>1.7.20</slf4j.version>
  11.     </properties>
  12.  
  13.     <dependencies>
  14.         <dependency>
  15.             <groupId>org.scala-lang</groupId>
  16.             <artifactId>scala-library</artifactId>
  17.             <version>${scala.version}</version>
  18.         </dependency>
  19.         <dependency>
  20.             <groupId>org.apache.spark</groupId>
  21.             <artifactId>spark-core_2.11</artifactId>
  22.             <version>${spark.version}</version>
  23.             <scope>provided</scope>
  24.         </dependency>
  25.         <dependency>
  26.             <groupId>org.apache.spark</groupId>
  27.             <artifactId>spark-sql_2.11</artifactId>
  28.             <version>${spark.version}</version>
  29.             <scope>provided</scope>
  30.         </dependency>
  31.         <dependency>
  32.             <groupId>org.apache.spark</groupId>
  33.             <artifactId>spark-hive_2.11</artifactId>
  34.             <version>${spark.version}</version>
  35.             <scope>provided</scope>
  36.         </dependency>
  37.         <dependency>
  38.             <groupId>org.apache.hadoop</groupId>
  39.             <artifactId>hadoop-client</artifactId>
  40.             <version>${hadoop.version}</version>
  41.             <scope>provided</scope>
  42.         </dependency>
  43.         <dependency>
  44.             <groupId>org.apache.hadoop</groupId>
  45.             <artifactId>hadoop-common</artifactId>
  46.             <version>${hadoop.version}</version>
  47.             <scope>provided</scope>
  48.         </dependency>
  49.         <dependency>
  50.             <groupId>redis.clients</groupId>
  51.             <artifactId>jedis</artifactId>
  52.             <version>3.1.0</version>
  53.         </dependency>
  54.     </dependencies>
  55.  
  56.     <build>
  57.         <pluginManagement>
  58.             <plugins>
  59.                 <!-- 该插件用于将Scala代码编译成class文件 -->
  60.                 <plugin>
  61.                     <groupId>net.alchim31.maven</groupId>
  62.                     <artifactId>scala-maven-plugin</artifactId>
  63.                     <version>3.2.0</version>
  64.                 </plugin>
  65.                 <plugin>
  66.                     <groupId>org.apache.maven.plugins</groupId>
  67.                     <artifactId>maven-compiler-plugin</artifactId>
  68.                     <version>3.8.1</version>
  69.                 </plugin>
  70.             </plugins>
  71.         </pluginManagement>
  72.         <plugins>
  73.             <plugin>
  74.                 <groupId>net.alchim31.maven</groupId>
  75.                 <artifactId>scala-maven-plugin</artifactId>
  76.                 <executions>
  77.                     <execution>
  78.                         <id>scala-compile-first</id>
  79.                         <phase>process-resources</phase>
  80.                         <goals>
  81.                             <goal>add-source</goal>
  82.                             <goal>compile</goal>
  83.                         </goals>
  84.                     </execution>
  85.                     <execution>
  86.                         <id>scala-test-compile</id>
  87.                         <phase>process-test-resources</phase>
  88.                         <goals>
  89.                             <goal>testCompile</goal>
  90.                         </goals>
  91.                     </execution>
  92.                 </executions>
  93.             </plugin>
  94.             <plugin>
  95.                 <groupId>org.apache.maven.plugins</groupId>
  96.                 <artifactId>maven-compiler-plugin</artifactId>
  97.                 <executions>
  98.                     <execution>
  99.                         <phase>compile</phase>
  100.                         <goals>
  101.                             <goal>compile</goal>
  102.                         </goals>
  103.                     </execution>
  104.                 </executions>
  105.             </plugin>
  106.             <plugin>
  107.                 <groupId>org.apache.maven.plugins</groupId>
  108.                 <artifactId>maven-shade-plugin</artifactId>
  109.                 <version>2.4.3</version>
  110.                 <executions>
  111.                     <execution>
  112.                         <phase>package</phase>
  113.                         <goals>
  114.                             <goal>shade</goal>
  115.                         </goals>
  116.                         <configuration>
  117.                             <filters>
  118.                                 <filter>
  119.                                     <artifact>*:*</artifact>
  120.                                     <excludes>
  121.                                         <exclude>META-INF/*.SF</exclude>
  122.                                         <exclude>META-INF/*.DSA</exclude>
  123.                                         <exclude>META-INF/*.RSA</exclude>
  124.                                     </excludes>
  125.                                 </filter>
  126.                             </filters>
  127.                         </configuration>
  128.                     </execution>
  129.                 </executions>
  130.             </plugin>
  131.         </plugins>
  132.     </build>

 Redis

RDB文件 

结果 

  1. hash 5 {37, 41, 61, 79, 83}		256m
  2.  
  3. 原表		    导入		    差值
  4. 17440258	17440227	31	
  5. 6675384		6675384		0	
  6. 11730951	11730951	0	
  7. 3489212		3489212		0	
  8.  
  9. 总数
  10. 39,335,805‬	39,335,774‬      31
声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/花生_TL007/article/detail/462677
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号