public class BitMap {  
    privatefinalbyte[] bits;  
    
    public BitMap(int maxNum) {  
        this.bits = newbyte[(maxNum >> 3) + 1]; // 每byte存儲(chǔ)8個(gè)數(shù)字  
    }  
    
    public void add(int num) {  
        int arrayIndex = num >> 3;  // num/8  
        int position = num & 0x07;  // num%8  
        bits[arrayIndex] |= 1 << position;  
    }  
    
    public boolean contains(int num) {  
        int arrayIndex = num >> 3;  
        int position = num & 0x07;  
        return (bits[arrayIndex] & (1 << position)) != 0;  
    }  
}

(10^10 - 10^4) / 8 / 1024/1024 ≈ 1.16GB

// 偏移量?jī)?yōu)化：存儲(chǔ)(qq - 10000)  
public void add(long qq) {  
    long num = qq - 10000;  
    int arrayIndex = (int)(num >> 3);  
    int position = (int)(num & 7);  
    bits[arrayIndex] |= 1 << position;  
}

public class BloomFilter {  
    privatefinal BitSet bitset;  
    privatefinalint size;  
    privatefinalint[] seeds;  
    
    public BloomFilter(int size, int hashCount) {  
        this.bitset = new BitSet(size);  
        this.size = size;  
        this.seeds = newint[hashCount];  
        for (int i = 0; i < hashCount; i++) {  
            seeds[i] = i * 31;  
        }  
    }  
    
    public void add(String qq) {  
        for (int seed : seeds) {  
            int hash = hash(qq, seed);  
            bitset.set(Math.abs(hash % size), true);  
        }  
    }  
    
    public boolean contains(String qq) {  
        for (int seed : seeds) {  
            int hash = hash(qq, seed);  
            if (!bitset.get(Math.abs(hash % size))) {  
                returnfalse;  
            }  
        }  
        returntrue;  
    }  
    
    private int hash(String value, int seed) {  
        // MurmurHash 實(shí)現(xiàn)  
        int result = 0;  
        for (char c : value.toCharArray()) {  
            result = seed * result + c;  
        }  
        return result;  
    }  
}

// 外部排序  
public void externalSort(String input, String output) throws IOException {  
    List<File> chunks = splitAndSort(input, 100_000_000); // 每個(gè)文件1千萬(wàn)  
    mergeFiles(chunks, output);  
}  

// 多路歸并  
void mergeFiles(List<File> files, String output) {  
    PriorityQueue<MergeEntry> queue = new PriorityQueue<>();  
    List<BufferedReader> readers = new ArrayList<>();  

    // 初始化堆  
    for (File file : files) {  
        BufferedReader reader = new BufferedReader(new FileReader(file));  
        readers.add(reader);  
        String line = reader.readLine();  
        if (line != null) {  
            queue.add(new MergeEntry(line, reader));  
        }  
    }  

    try (BufferedWriter writer = new BufferedWriter(new FileWriter(output))) {  
        long last = -1;  
        while (!queue.isEmpty()) {  
            MergeEntry entry = queue.poll();  
            long qq = Long.parseLong(entry.value);  

            // 去重：只寫入不重復(fù)的QQ號(hào)  
            if (qq != last) {  
                writer.write(entry.value);  
                writer.newLine();  
                last = qq;  
            }  

            // 讀取下一行  
            String next = entry.reader.readLine();  
            if (next != null) {  
                queue.add(new MergeEntry(next, entry.reader));  
            }  
        }  
    } finally {  
        readers.forEach(r -> { try { r.close(); } catch (IOException e) {}});  
    }  
}  

class MergeEntry implements Comparable<MergeEntry> {  
    String value;  
    BufferedReader reader;  

    public MergeEntry(String value, BufferedReader reader) {  
        this.value = value;  
        this.reader = reader;  
    }  

    @Override
    public int compareTo(MergeEntry o) {  
        returnthis.value.compareTo(o.value);  
    }  
}

val qqRDD = spark.read.textFile("hdfs://qq_data/*.txt")  
  .map(_.toLong)  
  .repartition(1000)  // 分為1000個(gè)分區(qū)  

// 每個(gè)分區(qū)內(nèi)部去重  
val distinctRDD = qqRDD.mapPartitions { iter =>  
val bitmap = newRoaringBitmap()  
  iter.foreach(qq => bitmap.add(qq.toInt))  
  bitmap.iterator.asScala.map(_.toLong)  
}  

// 全局去重（可選）  
val globalDistinct = distinctRDD.distinct()  

globalDistinct.saveAsTextFile("hdfs://result/")

普通位圖：10^10 / 8 / 1024/1024 ≈ 1.16 GB  
RoaringBitmap：稀疏數(shù)據(jù)下可壓縮至100-300 MB

public class QQDeduplication {  
    privatestaticfinal String REDIS_KEY = "qq_set";  

    public boolean isDuplicate(String qq) {  
        try (Jedis jedis = jedisPool.getResource()) {  
            // 使用HyperLogLog進(jìn)行基數(shù)估計(jì)  
            if (jedis.pfcount(REDIS_KEY) > 1_000_000_000L) {  
                returntrue; // 已超過(guò)10億，直接返回重復(fù)  
            }  
            return jedis.sadd(REDIS_KEY, qq) == 0;  
        }  
    }  
}

100 × 100 × 1.2MB = 12GB（分布式存儲(chǔ)可行）

public class LayeredBitmap {  
    privatefinal RoaringBitmap[][][] bitmaps;  
    privatestaticfinalint L1 = 100; // 一級(jí)分片  
    privatestaticfinalint L2 = 100; // 二級(jí)分片  

    public LayeredBitmap() {  
        bitmaps = new RoaringBitmap[L1][L2][];  
    }  

    public void add(long qq) {  
        int l1Index = (int)((qq - 10000) / 100_000_000);  
        long remainder = (qq - 10000) % 100_000_000;  
        int l2Index = (int)(remainder / 1_000_000);  
        int value = (int)(remainder % 1_000_000);  

        if (bitmaps[l1Index][l2Index] == null) {  
            bitmaps[l1Index][l2Index] = new RoaringBitmap();  
        }  
        bitmaps[l1Index][l2Index].add(value);  
    }  

    public boolean contains(long qq) {  
        // 類似add的分片計(jì)算  
        RoaringBitmap bitmap = bitmaps[l1Index][l2Index];  
        return bitmap != null && bitmap.contains(value);  
    }  
}

// 動(dòng)態(tài)分片函數(shù)  
int getShardId(long qq, int shardCount) {  
    String str = String.valueOf(qq);  
    // 取后6位作為分片依據(jù)  
    int suffix = Integer.parseInt(str.substring(Math.max(0, str.length() - 6)));  
    return suffix % shardCount;  
}