Uber Search
Platform

import threading, queue, time
from dataclasses import dataclass
from kafka import KafkaConsumer

@dataclass
class Message:
    offset: int
    key: str
    payload: dict

class StreamPoller:
    """Kafka에서 자체 페이스로 끌어와 BlockingQueue로 흘리는 worker.
    backpressure는 큐가 full일 때 자연 발생 — Kafka는 그대로 대기."""

    def __init__(self, topic: str, shard_id: int, queue_size: int = 1000):
        self.consumer = KafkaConsumer(topic, group_id=f"os-shard-{shard_id}",
                                      auto_offset_reset="earliest",
                                      enable_auto_commit=False)
        self.queue = queue.Queue(maxsize=queue_size)
        self.start_pointer = 0      # BatchStartPointer
        self.in_flight_offsets = set()

    def poll_loop(self):
        for msg in self.consumer:
            # 큐가 full이면 put이 block → poller도 block
            # → Kafka에서 더 안 받음 (= 자연 backpressure)
            self.queue.put(Message(msg.offset, msg.key.decode(), msg.value))
            self.in_flight_offsets.add(msg.offset)

    def commit(self, processed_offsets: set):
        # BatchStartPointer = 모든 worker가 처리 완료한 최소 offset
        self.in_flight_offsets -= processed_offsets
        new_start = (min(self.in_flight_offsets)
                     if self.in_flight_offsets else max(processed_offsets) + 1)
        self.start_pointer = new_start
        self.consumer.commit({...: new_start})

import schedule, time
from typing import List
from lucene import IndexWriter, DirectoryReader

class BSLIndex:
    """Base / Snapshot / Live 3-layer 인덱스 — Sia 패턴 재현."""

    def __init__(self):
        self.live = InMemoryLuceneIndex()      # 메모리 NRT
        self.snapshots: List[DirectoryReader] = []
        self.base: DirectoryReader = None

    # ─── Layer 1: Live (always write here) ───
    def index_document(self, doc: dict):
        self.live.add(doc)                    # ms 단위 가시성

    # ─── Layer 2: ~30min마다 Live → Snapshot flush ───
    def flush_live_to_snapshot(self):
        snapshot = self.live.flush_to_disk()   # 표준 Lucene segment
        self.snapshots.append(snapshot)
        self.live.reset()                    # 메모리 해제

    # ─── Layer 3: weekly snapshot들 → Base compaction ───
    def build_base(self):
        # 누적된 snapshot들 + 기존 base를 force-merge
        merged = force_merge([self.base] + self.snapshots)
        self.base = merged
        self.snapshots = []                # 컴팩션 후 비움

    # ─── 쿼리: 3개 layer 모두 보고 union ───
    def search(self, query: str) -> list:
        results = []
        results += self.live.search(query)                # NRT
        for snap in self.snapshots:
            results += snap.search(query)                  # recent
        if self.base:
            results += self.base.search(query)              # historical
        return merge_topk(results)

# 스케줄러 (Sia는 cron으로 운영했을 것)
schedule.every(30).minutes.do(idx.flush_live_to_snapshot)
schedule.every().week.do(idx.build_base)

from pyspark.sql import SparkSession

spark = (SparkSession.builder
    .appName("vector-bulk-ingest")
    .config("spark.executor.cores", 16)         # 충분한 cores
    .config("spark.executor.instances", 100)    # 병렬 폭
    .config("spark.sql.shuffle.partitions", 800)
    .getOrCreate())

# 핵심: 인덱스 설정을 bulk-friendly로 — Uber가 강조한 값들
INDEX_SETTINGS = {
    "index": {
        "refresh_interval": "-1",            # 1s → off, 요청별 refresh
        "translog": {
            "flush_threshold_size": "1024mb",    # 518 → 1024 MB
            "sync_interval": "120s",            # 매 요청 → 120s
        },
        "merge": {
            "policy": {
                "floor_segment": "10m",           # 1m → 10m
                "segments_per_tier": 15,            # 10 → 15
                "max_merge_at_once": 15,            # 10 → 15
            }
        },
        "knn.algo_param.index_thread_qty": 8, # native KNN 빌드 스레드
    },
    "mappings": {
        "_source": {"enabled": False},        # 인덱스 크기 -50%+
        "properties": {
            "id": {"type": "keyword", "doc_values": False},
            "vec": {
                "type": "knn_vector", "dimension": 400,
                "method": {"name": "hnsw", "engine": "faiss"},
            }
        }
    }
}

(spark.read.table("hive.embeddings_table")
   .repartition(800)
   .write
   .format("opensearch")               # spark-opensearch-connector
   .option("opensearch.resource", "vec-v2-green")
   .option("opensearch.batch.size.bytes", "50mb")
   .option("opensearch.batch.write.refresh", "false")
   .mode("append")
   .save())

# 빌드 완료 후 명시적 force_merge → 단일 segment
os_client.indices.forcemerge(index="vec-v2-green", max_num_segments=1)

from opensearchpy import OpenSearch

client = OpenSearch([{"host": "prod.cluster", "port": 9200}])

ALIAS = "vec-current"             # 쿼리는 항상 alias로

def deploy_green(green_index: str):
    """새로 빌드한 green 인덱스를 검증 후 traffic 스왑."""

    # ─── Gate 1: 완전성 (doc count가 비슷한가) ───
    blue = resolve_current_index(ALIAS)
    blue_count  = client.count(index=blue)["count"]
    green_count = client.count(index=green_index)["count"]
    assert green_count >= blue_count * 0.99, \
        f"green {green_count} too small vs blue {blue_count}"

    # ─── Gate 2: 매핑 하위 호환 (필드 추가만, 삭제·타입변경 금지) ───
    assert mapping_compatible(blue, green_index), "mapping breaking change"

    # ─── Gate 3: recall regression (sample 쿼리 결과 비교) ───
    for q in load_eval_queries(n=100):
        r_blue  = client.search(index=blue,        body=q)["hits"]["hits"]
        r_green = client.search(index=green_index, body=q)["hits"]["hits"]
        recall_at_10 = overlap([h["_id"] for h in r_blue[:10]],
                              [h["_id"] for h in r_green[:10]])
        assert recall_at_10 > 0.95, f"recall regression on {q}"

    # ─── 원자적 swap — 한 줄, downtime 0 ───
    client.indices.update_aliases(body={
        "actions": [
            {"remove": {"index": blue,        "alias": ALIAS}},
            {"add":    {"index": green_index, "alias": ALIAS}},
        ]
    })

    # 24h 후 blue 인덱스 삭제 (rollback 여유)
    schedule_delete(blue, after="24h")

import grpc
from opensearch_pb2 import BulkRequest, IndexAction, Document
from opensearch_pb2_grpc import BulkServiceStub

# OpenSearch gRPC endpoint — Uber가 contribute한 native API
channel = grpc.secure_channel("opensearch.internal:9300",
                              grpc.ssl_channel_credentials())
bulk_stub = BulkServiceStub(channel)

def bulk_index(docs: list[dict], index: str):
    """REST/JSON 거치지 않고 Protobuf 그대로 — translation layer 0."""

    actions = []
    for d in docs:
        doc_pb = Document(
            id=d["id"],
            # vector는 Protobuf의 packed repeated float — bytes 효율 88% 절감
            vector=d["vec"],
            metadata={k: v for k, v in d.items() if k not in ["id", "vec"]},
        )
        actions.append(IndexAction(index=index, document=doc_pb))

    request = BulkRequest(actions=actions, refresh=False)
    response = bulk_stub.Bulk(request, timeout=30.0)

    if response.errors:
        for item in response.items:
            if item.error:
                log_error(item.id, item.error.message)
    return response.took_ms

# Spark batch job에서 사용: runtime -20~35%
# M3 metrics: p99 write 34.1ms → 13.6ms (-60%)
# Delivery vector: p50 83ms → 38ms (-53%)