如何统计 UV、独立访客与海量数据去重？

@Service
@Slf4j
public class UvService {

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    private static final String UV_KEY_PREFIX = "uv:daily:";

    /**
     * 记录一次访问
     * userId 可以是登录用户 ID，也可以是匿名访客的 cookie/session ID
     */
    public void recordVisit(String userId) {
        LocalDate today = LocalDate.now();
        String key = UV_KEY_PREFIX + today;
        // PFADD 自动去重
        redisTemplate.opsForHyperLogLog().add(key, userId);
    }

    /**
     * 获取当天 DAU
     */
    public long getDailyUv() {
        String key = UV_KEY_PREFIX + LocalDate.now();
        return redisTemplate.opsForHyperLogLog().size(key);
    }

    /**
     * 获取时间段内 UV（去重合并）
     * 例：过去 7 天的去重 UV
     */
    public long getUvBetween(LocalDate start, LocalDate end) {
        List<String> keys = new ArrayList<>();
        for (LocalDate d = start; !d.isAfter(end); d = d.plusDays(1)) {
            keys.add(UV_KEY_PREFIX + d.toString());
        }
        if (keys.isEmpty()) return 0;

        return redisTemplate.opsForHyperLogLog().size(
            keys.toArray(new String[0])
        );
    }

    /**
     * 设置过期时间（保留 90 天）
     */
    @Scheduled(cron = "0 5 0 * * ?") // 每天凌晨 00:05 执行
    public void setKeyExpire() {
        LocalDate today = LocalDate.now();
        String key = UV_KEY_PREFIX + today;
        redisTemplate.expire(key, 90, TimeUnit.DAYS);
    }
}

@Service
@Slf4j
public class UvBitmapService {

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    /**
     * 用户访问 → 对应 bit 位置 1
     */
    public void recordVisit(long userId) {
        LocalDate today = LocalDate.now();
        String key = String.format("uv:bitmap:%s", today);
        // userId - 1，因为 bit 从 0 开始
        redisTemplate.opsForValue().setBit(key, userId - 1, true);
    }

    /**
     * 获取当天 DAU
     */
    public long getDailyUv() {
        String key = String.format("uv:bitmap:%s", LocalDate.now());
        return redisTemplate.opsForValue().bitCount(key);
    }

    /**
     * 获取指定用户今天是否访问
     */
    public boolean hasVisited(long userId) {
        String key = String.format("uv:bitmap:%s", LocalDate.now());
        return Boolean.TRUE.equals(
            redisTemplate.opsForValue().getBit(key, userId - 1)
        );
    }

    /**
     * 跨天 UV 合并（BITOP OR）
     */
    public long getUvBetween(LocalDate start, LocalDate end) {
        String destKey = String.format("uv:bitmap:merged:%s_%s", start, end);
        List<String> srcKeys = new ArrayList<>();
        for (LocalDate d = start; !d.isAfter(end); d = d.plusDays(1)) {
            srcKeys.add(String.format("uv:bitmap:%s", d));
        }

        // BITOP OR：将所有天的 bitmap OR 合并，结果存 destKey
        redisTemplate.opsForValue().bitOp(
            RedisStringCommands.BitOperation.OR,
            destKey,
            srcKeys.toArray(new String[0])
        );

        long uv = redisTemplate.opsForValue().bitCount(destKey);
        // 合并完删除临时 key
        redisTemplate.delete(destKey);
        return uv;
    }
}

public class ExternalMergeSort {

    private static final long BATCH_SIZE = 50_000_000L; // 5000 万条

    /**
     * 第一阶段：分批排序写文件
     */
    public void phase1Sort(String inputFile, List<String> tempFiles) throws IOException {
        long totalRecords = countRecords(inputFile);
        long batches = (totalRecords + BATCH_SIZE - 1) / BATCH_SIZE;

        try (FileInputStream fis = new FileInputStream(inputFile);
             BufferedInputStream bis = new BufferedInputStream(fis, 8 * 1024 * 1024);
             DataInputStream dis = new DataInputStream(bis)) {

            for (long b = 0; b < batches; b++) {
                long[] batch = new long[(int) Math.min(BATCH_SIZE, totalRecords - b * BATCH_SIZE)];
                for (int i = 0; i < batch.length; i++) {
                    batch[i] = dis.readLong();
                }
                Arrays.sort(batch);
                String tempFile = "/tmp/sorted_" + b + ".bin";
                writeSortedBatch(tempFile, batch);
                tempFiles.add(tempFile);
            }
        }
    }

    /**
     * 第二阶段：N 路归并去重
     */
    public void phase2Merge(List<String> tempFiles, String outputFile) throws IOException {
        // 1. 初始化 N 个文件的迭代器
        PriorityQueue<RecordPointer> heap = new PriorityQueue<>();
        List<DataInputStream> inputs = new ArrayList<>();

        for (int i = 0; i < tempFiles.size(); i++) {
            DataInputStream dis = new DataInputStream(
                new BufferedInputStream(new FileInputStream(tempFiles.get(i)), 1024 * 1024));
            inputs.add(dis);
            if (dis.available() > 0) {
                heap.add(new RecordPointer(i, dis.readLong()));
            }
        }

        // 2. 归并输出
        try (BufferedOutputStream bos = new BufferedOutputStream(
                new FileOutputStream(outputFile), 8 * 1024 * 1024);
             DataOutputStream dos = new DataOutputStream(bos)) {

            long last = -1;
            while (!heap.isEmpty()) {
                RecordPointer min = heap.poll();
                long current = min.value;

                // 去重：只输出与上一个不同的
                if (current != last) {
                    dos.writeLong(current);
                    last = current;
                }

                // 从同一文件读下一条
                DataInputStream dis = inputs.get(min.fileIndex);
                if (dis.available() >= 8) {
                    min.value = dis.readLong();
                    heap.add(min);
                }
            }
        }
    }
}

/**
 * Flink 实时 DAU 统计（Kafka → Redis）
 */
public class DailyUvJob {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> events = env
            .addSource(new FlinkKafkaConsumer<>("user_visit", new JsonSchema(), props))
            .uid("kafka-source");

        events.keyBy(e -> e.getDate())  // 按天分组
            .window(SlidingEventTimeWindows.of(Time.hours(1), Time.hours(1)))
            .apply(new UvAggregator())
            .addSink(new RedisSink<>()); // 结果写入 Redis

        env.execute("Daily UV Job");
    }

    // 聚合函数：PFADD
    public static class UvAggregator
            implements WindowFunction<String, Long, String, TimeWindow> {

        @Override
        public void apply(String key, TimeWindow window,
                         Iterable<String> records, Collector<Long> out) {
            RedisCommands<String, String> redis = jedisPool.getResource();
            String hllKey = "uv:daily:" + key;
            for (String userId : records) {
                redis.pfadd(hllKey, userId);
            }
            long uv = redis.pfcount(hllKey);
            out.collect(uv);
        }
    }
}