征服大数据算法面试：核心题型与实战解析

import heapq
from collections import defaultdict

def top_k_frequent(nums, k):
    # 统计频率 O(n)
    freq_map = defaultdict(int)
    for num in nums:
        freq_map[num] += 1
    
    # 维护最小堆 O(nlogk)
    heap = []
    for num, freq in freq_map.items():
        if len(heap) < k:
            heapq.heappush(heap, (freq, num))
        else:
            if freq > heap[0][0]:
                heapq.heappop(heap)
                heapq.heappush(heap, (freq, num))
    
    return [num for (freq, num) in heap]

# Mapper阶段
def mapper(chunk):
    local_count = {}
    for num in chunk:
        local_count[num] = local_count.get(num, 0) + 1
    yield from local_count.items()

# Reducer阶段
def reducer(results, k):
    global_count = {}
    for num, cnt in results:
        global_count[num] = global_count.get(num, 0) + cnt
    
    # 使用堆获取TopK
    return heapq.nlargest(k, global_count.items(), key=lambda x: x[1])

import random

def reservoir_sampling(stream, k):
    reservoir = []
    for i, item in enumerate(stream):
        if i < k:
            reservoir.append(item)
        else:
            j = random.randint(0, i)
            if j < k:
                reservoir[j] = item
    return reservoir

import hashlib
import math

class HyperLogLog:
    def __init__(self, precision=12):
        self.p = precision
        self.m = 1 << precision
        self.registers = [0] * self.m
    
    def add(self, element):
        hashed = int(hashlib.md5(element.encode()).hexdigest(), 16)
        index = hashed & (self.m - 1)
        w = hashed >> self.p
        self.registers[index] = max(self.registers[index], 
                                  (w & -w).bit_length())
    
    def count(self):
        harmonic_mean = sum(2**-r for r in self.registers)
        estimate = self.m * self.m * 0.7213 / harmonic_mean
        return int(estimate)

import hashlib

class ConsistentHash:
    def __init__(self, nodes, replica=3):
        self.replica = replica
        self.ring = {}
        for node in nodes:
            for i in range(replica):
                key = self._hash(f"{node}:{i}")
                self.ring[key] = node
        self.sorted_keys = sorted(self.ring.keys())
    
    def _hash(self, key):
        return int(hashlib.md5(key.encode()).hexdigest(), 16)
    
    def get_node(self, key):
        if not self.ring:
            return None
        hash_key = self._hash(key)
        idx = bisect.bisect(self.sorted_keys, hash_key) % len(self.sorted_keys)
        return self.ring[self.sorted_keys[idx]]

import mmh3
from bitarray import bitarray

class BloomFilter:
    def __init__(self, size, hash_num):
        self.size = size
        self.hash_num = hash_num
        self.bit_array = bitarray(size)
        self.bit_array.setall(0)
    
    def add(self, item):
        for seed in range(self.hash_num):
            index = mmh3.hash(item, seed) % self.size
            self.bit_array[index] = 1
    
    def contains(self, item):
        for seed in range(self.hash_num):
            index = mmh3.hash(item, seed) % self.size
            if not self.bit_array[index]:
                return False
        return True

class BitMap:
    def __init__(self, max_num):
        self._size = (max_num + 7) // 8
        self.bitmap = bytearray(self._size)
    
    def set(self, num):
        index = num // 8
        offset = num % 8
        self.bitmap[index] |= 1 << offset
    
    def get(self, num):
        index = num // 8
        offset = num % 8
        return (self.bitmap[index] & (1 << offset)) != 0