라이브코딩 테스트: 블룸 필터

블룸 필터와 확률적 자료구조에 대한 이해도, 메모리 효율성 인식, 실무 문제 해결 능력을 평가하기 위한 질문입니다.

1단계: 블룸 필터 버그 수정

아래 블룸 필터 구현에 버그가 있습니다. 테스트 결과를 보면 False Positive Rate가 비정상적으로 높습니다

  • 버그가 무엇인지 찾고 수정해주세요. 왜 이것이 버그인지 설명해주세요.
import hashlib

class BloomFilter:

    def __init__(self, size: int, hash_count: int):
        self.size = size
        self.hash_count = hash_count
        self.bit_array = [False] * size

    def _hashes(self, item: str) -> list[int]:
        result = []

        for i in range(self.hash_count):
            data = f"{item}:{i}"
            h = hashlib.sha256(data.encode()).hexdigest()
            idx = int(h, 16) % self.size
            result.append(idx)

        return result

    def add(self, item: str) -> None:
        for idx in self._hashes(item):
            self.bit_array[idx] = True

    def contains(self, item: str) -> bool:
        return any(self.bit_array[idx] for idx in self._hashes(item))


if __name__ == "__main__":
    bf = BloomFilter(size=10000, hash_count=5)

    for i in range(100):
        bf.add(f"user_{i}")

    added_check = all(bf.contains(f"user_{i}") for i in range(100))
    print(f"추가한 유저: {added_check}")
    # 예상: True

    not_added = [bf.contains(f"visitor_{i}") for i in range(1000)]
    false_positive_rate = sum(not_added) / len(not_added)
    print(f"False Positive Rate: {false_positive_rate:.2%}")
    # 예상: 약 0.01%
    # 실제: 약 25% ← 왜 이렇게 높을까요?

2단계: Counting Bloom Filter 확장

1단계에서 수정한 블룸 필터는 잘 동작하지만 삭제가 안됩니다. 비트 대신 숫자를 사용하는 Counting Bloom Filter를 구현합니다

  • contains remove 메소드를 구현해주세요
import hashlib


class CountingBloomFilter:

    def __init__(self, size: int, hash_count: int):
        self.size = size
        self.hash_count = hash_count
        self.count_array = [0] * size

    def _hashes(self, item: str) -> list[int]:
        result = []

        for i in range(self.hash_count):
            data = f"{item}:{i}"
            h = hashlib.sha256(data.encode()).hexdigest()
            idx = int(h, 16) % self.size
            result.append(idx)

        return result

    def add(self, item: str) -> None:
        for idx in self._hashes(item):
            self.count_array[idx] += 1

    def contains(self, item: str) -> bool:
        """아이템이 블룸 필터에 존재하는지 확인합니다."""
        # TODO: 구현하세요
        pass

    def remove(self, item: str) -> bool:
        """
        아이템을 블룸 필터에서 제거합니다.

        Returns:
            True: 삭제 성공
            False: 삭제 실패 (존재하지 않는 아이템)

        주의: 카운터가 음수가 되면 안 됩니다.
        """
        # TODO: 구현하세요
        pass


if __name__ == "__main__":
    cbf = CountingBloomFilter(size=10000, hash_count=5)

    # 테스트 1: 추가 후 확인
    cbf.add("user_1")
    cbf.add("user_2")
    print(f"user_1 존재: {cbf.contains('user_1')}")  # 예상: True
    print(f"user_2 존재: {cbf.contains('user_2')}")  # 예상: True

    # 테스트 2: 삭제 후 확인
    cbf.remove("user_1")
    print(f"user_1 존재 (삭제 후): {cbf.contains('user_1')}")  # 예상: False
    print(f"user_2 존재 (user_1 삭제 후): {cbf.contains('user_2')}")  # 예상: True

    # 테스트 3: 같은 아이템 여러 번 추가
    cbf.add("user_3")
    cbf.add("user_3")
    cbf.remove("user_3")
    print(f"user_3 존재 (2번 추가, 1번 삭제): {cbf.contains('user_3')}")  # 예상: True

    # 테스트 4: 존재하지 않는 아이템 삭제
    result = cbf.remove("user_999")
    print(f"존재하지 않는 아이템 삭제 결과: {result}")  # 예상: False

3단계: 실무 적용 - 중복 푸시 알림 방지

100만명의 유저에게 프로모션 푸시 알림을 발송하려고 합니다. 발송 시스템이 불안정해서 중간에 재시작될 수 있고, 재시작 후 처음부터 다시 발송하면 일부 유저가 중복 수신하는 문제가 있습니다.

  • send_push 메서드를 완성해주세요
  • 제약 조건
    • 대상 유저: 1,000,000명
    • 허용 가능한 False Positive Rate: 1% 이하
    • 메모리 제한: 블룸 필터는 최대 2MB 이하
import hashlib
import sys

from bitarray import bitarray


class BloomFilter:

    def __init__(self, size: int, hash_count: int):
        self.size = size
        self.hash_count = hash_count
        self.bit_array = bitarray(size)
        self.bit_array.setall(0)

    def _hashes(self, item: str) -> list[int]:
        result = []

        for i in range(self.hash_count):
            data = f"{item}:{i}"
            h = hashlib.sha256(data.encode()).hexdigest()
            idx = int(h, 16) % self.size
            result.append(idx)

        return result

    def add(self, item: str) -> None:
        for idx in self._hashes(item):
            self.bit_array[idx] = 1

    def contains(self, item: str) -> bool:
        return all(self.bit_array[idx] for idx in self._hashes(item))

    def memory_usage(self) -> int:
        return sys.getsizeof(self.bit_array)


class PushService:

    def __init__(self):
        self.sent_filter = BloomFilter(size=10_000_000, hash_count=7)
        self.send_count = 0
        self.skip_count = 0

    def _send_push_notification(self, user_id: str, message: str) -> None:
        """실제 푸시 발송 (시뮬레이션)"""
        pass

    def send_push(self, user_id: str, message: str) -> bool:
        """
        푸시 알림을 발송합니다.

        Args:
            user_id: 유저 ID
            message: 푸시 메시지

        Returns:
            True: 발송 성공
            False: 이미 발송됨 (스킵)
        """
        # TODO: 구현하세요
        pass

    def get_stats(self) -> dict:
        return {"sent": self.send_count, "skipped": self.skip_count}


if __name__ == "__main__":
    service = PushService()

    # 1차 발송 (50만 명)
    for i in range(500_000):
        service.send_push(f"user_{i}", "프로모션 알림!")

    print(f"1차 발송 완료: {service.get_stats()}")
    # 예상: sent=500,000, skipped=0

    # 서버 재시작 후 2차 발송 (100만 명 전체)
    for i in range(1_000_000):
        service.send_push(f"user_{i}", "프로모션 알림!")

    print(f"2차 발송 완료: {service.get_stats()}")
    # 예상: sent=1,000,000, skipped=500,000
  • 논의 사항
    • False Positive가 발생하면 어떤 문제가 생기나요?
    • 블룸 필터 대신 다른 방법은 없을까요?