[大数据]redis-高级特性Bitmaps，HyperLogLog GEO

Redis高级数据结构

Bitmaps 位图

位图法，就是用一段01来组成的连续的集合，1bit=8位，适用于大量数据的处理，可以使用很小的内存来处理大量的数据，布隆过滤器本质上就是使用位图法加多个hash函数来实现的。

命令：

• setbit
• getbit

布隆过滤器

使用k个hash函数对元素值进行k次计算，得到k个hash值。
根据得到的hash值，在位数组中把下标的值置为1。
布隆过滤器判断存在的不一定存在，但判断不存在的一定不存在，因为即使使用k个hash，依然有hash冲突的几率。
m 位数组长度 n数据个数 p误判率。

基于redis的布隆过滤器实现

public class RedisBloomFilter {

    /**
     * 误判率
     */
    private double errRate;

    /**
     * 需要存入的元素数量
     */
    private int maxElementSize;

    /**
     * 位图的长度
     */
    private long initSize;

    /**
     * 所需要的hash次数
     */
    private int hashSize;

    @Autowired
    private JedisPool jedisPool;

    private final static String RS_FILTER_NS = "rs:bloomfilter:";
    private final Logger logger = LoggerFactory.getLogger(RedisBloomFilter.class);

    public RedisBloomFilter init(double errRate, int maxElementSize) {
        // 计算布隆过滤器的 总数量和 hash 的个数 initSize = -(maxElementSize*errRate / ln2^2)
        // hashSize = initSize / maxElementSize * ln2^2
        this.initSize = (long) -((maxElementSize * Math.log(errRate)) / (Math.log(2) * Math.log(2)));
        this.hashSize = Math.max(1, (int) Math.round((double) initSize / maxElementSize * Math.log(2)));
        return this;
    }

    /**
     * 根据元素计算出该元素在位图中的位置
     *
     * @param element
     * @return
     */
    public long[] getBitIndices(String element) {
        long[] indices = new long[hashSize];
        // 将传入的字符串通过google的Hashing 转换为 128位的hash值 长度为16的字节数组
        byte[] bytes = Hashing.murmur3_128().
                hashObject(element, Funnels.stringFunnel(Charset.forName("UTF-8"))).asBytes();
        // 64位
        long hash1 = Longs.fromBytes(bytes[7], bytes[6], bytes[5], bytes[4], bytes[3], bytes[2], bytes[1], bytes[0]);
        // 64位
        long hash2 = Longs.fromBytes(bytes[15], bytes[14], bytes[13], bytes[12], bytes[11], bytes[10], bytes[9], bytes[8]);
        /*用这两个函数的hash值来模拟多个函数产生的值*/
        long combinedHash = hash1;
        for (int i = 0; i < hashSize; i++) {
            indices[i] = (combinedHash & Long.MAX_VALUE) % initSize;
            combinedHash = combinedHash + hash2;
        }
        System.out.println(Arrays.toString(indices));
        return indices;
    }

    /**
     * 添加数据到redis bitmaps中
     *
     * @param key
     * @param element
     * @param expireSec
     */
    public void insert(String key, String element, int expireSec) {
        if (key == null || element == null) {
            throw new RuntimeException("键值均不能为空");
        }
        String actulKey = RS_FILTER_NS.concat(key);
        try (Jedis jedis = jedisPool.getResource()) {
            long[] bitIndices = this.getBitIndices(element);
            Pipeline pipelined = jedis.pipelined();
            for (long indice : bitIndices) {
                pipelined.setbit(actulKey, indice, true);
            }
            pipelined.syncAndReturnAll();
            jedis.expire(actulKey, expireSec);
        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }

    /**
     * 判断元素是否存在
     *
     * @param key
     * @param element
     * @return
     */
    public boolean mayExist(String key, String element) {
        if (key == null || element == null) {
            throw new RuntimeException("键值都不允许为空");
        }
        String autualKey = RS_FILTER_NS.concat(key);
        try (Jedis jedis = jedisPool.getResource()) {
            long[] bitIndices = this.getBitIndices(element);
            Pipeline pipelined = jedis.pipelined();
            for (long indice : bitIndices) {
                pipelined.getbit(autualKey, indice);
            }
            return !pipelined.syncAndReturnAll().contains(false);
        } catch (Exception ex) {
            ex.printStackTrace();
        }
        return false;
    }

    @Override
    public String toString() {
        return "RedisBloomFilter{" +
                "errRate=" + errRate +
                ", maxElementSize=" + maxElementSize +
                ", initSize=" + initSize +
                ", hashSize=" + hashSize +
                '}';
    }
}

HyperLogLog

底层基于字符串来实现，基于位数组。用来解决统计UV数据这种问题，统计百万级的用户，一条大约占据15k大小，可以看到占用内存是很少的。UV和PV，PV指页面统计量，UV，用户对页面的方法统计，一个用户不管访问多少次这个页面算1次。不精确的去重计数方案，标准误差是0.81%，这样的精确度已经可以满足统计UV统计需求了。
命令:

• pfcount u:uid:1018 统计
• pfadd u:uid:1018 “u1” "u2"“u3” 添加数据
• pfmerge [deskey] [sourcekey…] 汇总计算，把多个统计汇总到一个中
  HyperLogLog是基于概率论中的伯努利实验结合了极大使然估算法，并做了分桶优化。

抛硬币
正面 和 反面
抛到一个正面算一个回合
最长的正面抛了多少次 kmax 最长的回合抛了多少次 n多少个回合 也就是n个回合，k的最大值
一共进行了几个回合 通过概率统计推出 n = 2^kmax
这种预估方法，在数据量比较小时误差是比较大的。
HyperLogLog 引入分桶概念
比如分为 100组，平均数，受概率的影响就比较小了，
当数据分布不均匀时，算法平均数受大数的影响就比较大了，hll中使用了调和平均数，也就是取它的倒数
算数平均数 A 1000 B 30000 算数平均数 15500
调和平均数 A 1000 B 30000 2/(1/1000+1/30000) 1953，调和平均数不太容易收到大数的影响，更加贴近实际情况，也就是大多数数据的大小。
HLL分桶后再平均
统计每个网页每天的UV数据
1.数据通过hash函数转化为比特串 hash xxx -> 101
2.分桶(大的位数组)
3.对应: id 来代表不同的用户， hash id -> 比特串 比如 对于一个比特串 10010000
把比特串的前两位拿出来 00 计算桶号 进入0号桶 剩下 100100 剩下比特串里面，1从右到左 3 也就是kmax = 3 ,把多个id生产的多个比特串按照上面方法放入分好的桶中，每个桶中的数据kmax是在变化的，然后根据极大似然估算法来进行统计。

Redis中的具体实现

大小 12kb
桶 16384个 2的14次方
每个桶 6个bit
pfadd 添加 每一个value值 转化为 64为bit串，前14位用来定位桶，剩余50个bit，统计kmax，比如极端情况下，在第50位出现了一个1,50 2进制 110010（kmax） ，就是要放入第n个桶里的数据,每个桶6个bit，极端情况下也是可以放入的。每个桶记录最大的 kmax。
16384 桶，每个桶kmax，当调用pfcount命令时，会按照调和平均数进行相关估算，同时 加以偏差修正。公式如下：

const* m* 根据数的大小进行的偏差修正。最终得到一个pfCount的统计值出来。
12kb 16384 * 6 / 8 / 1024= 12kb，也就是12kb用来统计 2的64次方个数据的统计，极大的节约空间。每个桶的访问次数有一个小小的约束。
使用也是基于redis命令使用，比较简单，这里就不提供示例代码了。

Redis 3.2版本提供 GEO 地理信息定位功能

地图原始的位置数据，使用二维的数据定位，经纬度，经度(-180,180],维度(-90,90]。业界比较通用的距离排序算法是GeoHash算法，Redis也是用GeoHash算法。基本原理也就是把二维的数据，通过hash算法映射到1维之上，如果在二维上两个点比较相近时，映射到1维上也比较相近。Redis使用52位整数来存放编码的，底层使用ZSET来实现，并不是一种单独的数据结构。
命令:

• geoadd [key] [longitude [latitude] [number] 增加坐标 longitude经度 latitude维度 number 名称
• geopos [key] [number] 获取坐标
• geodis [key] [number1] [number2] [m|km|ft|mi] 计算两个坐标的距离
• georadius [key] [longitude] [latitude] radius m|km|ft|mi [withdist] [isASC] redius 半径

代码实现:

public class RedisGEO {

    public final static String RS_GEO_NS = "rg:";

    @Autowired
    private JedisPool jedisPool;

    /**
     * 添加坐标
     *
     * @param key
     * @param longitide 经度
     * @param latitude  维度
     * @param member    名称
     * @return 映射的值
     */
    public long addLocation(String key, double longitide, double latitude, String member) {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            return jedis.geoadd(RS_GEO_NS + key, longitide, latitude, member);
        } catch (Exception ex) {
            throw new RuntimeException("添加坐标异常");
        } finally {
            jedis.close();
        }
    }

    /**
     * 最近的xx
     * @param key
     * @param member 距离哪个位置
     * @param radius 半径
     * @param withDist 是否返回数据中包含距离中心节点的距离
     * @param isASC 结果排序
     * @return
     */
    public List<GeoRadiusResponse> nearByMore(String key, String member, double radius, boolean withDist, boolean isASC){
        Jedis jedis = null;
        try{
            jedis = jedisPool.getResource();
            GeoRadiusParam geoRadiusParam = new GeoRadiusParam();
            if(withDist){
                geoRadiusParam.withDist();
            }
            if(isASC){
                geoRadiusParam.sortAscending();
            }else {
                geoRadiusParam.sortDescending();
            }
            List<GeoRadiusResponse> geoRadiusResponses = jedis.georadiusByMember(RS_GEO_NS + key, member, radius, GeoUnit.KM, geoRadiusParam);
            return geoRadiusResponses;
        }catch (Exception e){
            throw new RuntimeException("计算最近xxx失败");
        }finally {
            jedis.close();
        }
    }

}