Design YouTube Top K (HelloInterview) のメモ

この記事で学んだことをメモる。

High Level Design

• Tumbling window であるため、monthly window を monthly の頻度で作るので良い。

以降、Cascade Aggregation を用いた方法 (rollup aggregation) と、Flinkを用いた方法をまとめる。記事では、Cascade Aggregation は使っていない。

DD How can we cut down on the number of queries to the database?

ベストプラクティスに従い、write-through-update strategyを用いある。

📝 2026年現在のシステム設計におけるベストプラクティス
一般的なWebアプリケーション（CRUD系）の教科書では「データの整合性を保つためにキャッシュは消せ（Delete）」と教わります。しかし、高スループットな集計・ランキングシステムにおいては、キャッシュは「消すもの」ではなく「バックグラウンドで常に最新状態に更新し続けるもの（Warm Cache）」 というのが鉄則です。

TopK Redis Cache を集計済みテーブル (PerHourCounts テーブル等) の前に起き、read-through キャッシュとして用いる。TopK Cron job を用いて、window ごとの集計結果を出し、これを TopK Redis Cache に上書きつまり、write-through-update する。

また、TTL を少し長めに設定する (PerHourCounts キャッシュなら3時間等)。これにより、集計ジョブが失敗/遅延した場合でも、エラーを返す代わりに、一定期間は古い集計データを返すことができる。

📝 write-through-delete を用いない理由は以下二つ。

• TopK Cron jobでは、集計済みDBのシャードから、並行にTopKをクエリし、それらをマージする、という処理を行う。つまり、キャッシュする内容は、single entity result ではなく、derived result　である。従って write-through-delete を行うと、キャッシュミスを起こしたプロセスが、この重い処理を行う必要があり、p99 latency が非常に高くなってしまう。
• write-through-delete を行うと、キャッシュがなくなった直後に、複数のクライアントが TopK Redis Cache にクエリを投げる Cache Stampede problem の対応が必要になるから。

ref. Scaling Reads - Application-Level Caching (Redis)

DD How can we handle the massive number of writes to the database? [HARD] 🔥🔥

700k rps の view event を裁くために、Sharding x Batching Ingestion を使用する。
前述の通り、記事の内容をベースにした Cascade Aggregation の方法を説明する。

Kafka  <------ MinuteAggregationJob ----> PerMinuteCount ...
^^^            ^^^                        table
partitioned    1 worker per
by video_id    1 partition

Kafka  <------ MinuteAggregationJob ----> PerMinuteCount ...
^^^            ^^^                        table
roundrobin     1 worker per
               1 partition

🟢 Great Approach: Two-Phase Aggregation

以下のように、Kafkaのキューを2段構成にする。
LocalPerMinJob と GlobalPerMinJob のそれぞれは、1分間のWindowでのイベントを集計し、1分ごとにテーブルに書き込む。

Kafka  <------ LocalPerMinJob ---->  Kafka   <---- GlobalPerMinJob---->  PerMinuteCount ...
^^^            ^^^                   ^^^           ^^^                   table
roundrobin     1 worker per          paritioned    1 worker per
               1 partition.          by video_id   1 partition

GlobalPerMinJob に書き出されるレコードは以下のスキーマを持つ。

{
  "video_id": "video_12345", 
  "time_bucket_minute": "2026-06-01T10:00:00Z", 
  "partial_count": 5000 
}

★ 全てのPer*Countテーブルにおいてシャーディングが必要

上記の方法だと、Jobの実行ごとに、そのwindowで閲覧されたビデオのユニーク数と同じだけのレコードを Per*Count テーブルへ書き込む必要がある。
これは、過負荷であるため、それぞれの Per*Countテーブルを含むDBをvideo_idによってシャーディングする(あるいは自動シャーディング機能を備えた分散データベースを使用する)必要がある。

TODO: 以上は、Kafka のパーティションキーを video_id から round robin に変更できる前提の議論。
もしパーティションキーをvideo_idから変更できない場合にどうするか考える。

Kafka <--- Flink ---> PerHourCount table  (with sharding by video_id)
                      PerDayCount table   (with sharding by video_id)
                      PerMonthCount table (with sharding by video_id)

DD How do we optimize our top K queries? 🔥

現在、定期ジョブによる Cascade Aggregation により、シャーディングされた Per*Count テーブルが出来上がっている。これらを使って、Cron Job で、window ごとの TopK を求める必要がある。

🟢 Great Approach: Merge K sorted lists in Cron Job

DD1 で述べた TopK Cron job において、各DBシャードのTopKレコードをクエリする。これをマージし、グローバルなTopKのリストを作成する。これを TopK Redis Cache に書きこむ (= write-through-updateする)。

マージは O(KlogN) 時間で可能。

public final class VideoTopKJob {
    private static final int TOP_K = 100;
    private static final List<WindowSpec> WINDOWS = List.of(
        new WindowSpec("last_hour", java.time.Duration.ofHours(1)),
        new WindowSpec("last_day", java.time.Duration.ofDays(1)),
        new WindowSpec("last_month", java.time.Duration.ofDays(30))
    );

    private VideoTopKJob() {
    }

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

        KafkaSource<VideoViewEvent> source = KafkaSource.<VideoViewEvent>builder()
            // ...

        DataStream<VideoViewEvent> events = env.fromSource(
            source,
            WatermarkStrategy.<VideoViewEvent>forBoundedOutOfOrderness(java.time.Duration.ofSeconds(30))
                .withTimestampAssigner((event, timestamp) -> event.eventTime()),
            "video-views-source"
        );

        DataStream<VideoWindowCount> windowCounts = events
            .name("video-views-with-watermarks")
            .keyBy(VideoViewEvent::videoId)
            .process(new RollingWindowAggregator(WINDOWS))
            .name("rolling-window-aggregator");

        windowCounts
            .keyBy(VideoWindowCount::window)
            .process(new TopKAggregator(TOP_K))
            .name("top-k-per-window")
            .addSink(new RedisSortedSetSink(/* */)
            .name("write-top-k-to-cache");

        env.execute("Video Top-K Sliding Window");
    }

    private static final class RollingWindowAggregator extends KeyedProcessFunction<String, VideoViewEvent, VideoWindowCount> {
      // ...
    }

    private static final class TopKAggregator extends KeyedProcessFunction<String, VideoWindowCount, TopKResult> {
        // ...
    }

    private static final class RedisSortedSetSink extends RichSinkFunction<TopKResult> {
        // ...
    }
}

DD What if we need to support sliding windows? 🔥

まず、仕様として、「hourly ranking を sliding window で即時更新することはユーザーにとって価値があるが、daily や monthly のランキングをそこまで即時反映する必要はない」とpush backする。

さらに、Hourly window においても、「毎秒でランキング更新を反映する必要はなく、1分ごとに更新されるので十分」と仕様を提案する。

ということで、以下、「1分ごとに更新される、Hourly の slidnig window」 を考える。

🟢 Good Solution: Shift tumbling window

Shardされた PerHourCount テーブルと、PerMinCount テーブルを利用する。

以下、K = 100 の場合で、例を出して説明する。
専属の SlidingTopK Job を1分おきに走らせる。この Jobの中では､

• 8:00 実行 -> PerHourCount テーブルから、7:00-8:00 の PerHourCount を取得し、top 500 を Redis Sliding TopK に保存する。
• 8:01 実行 -> Redis Sliding TopK のvideo_id 一覧を取得し、それぞれのvideo について、7:00-7:01 の　PerMinuteCount と、8:00-8:01 の　PerMinuteCount を取得する。既存のレコードを current - decrement + increment 値で更新する。
• 8:02 実行 -> 8:01と同様。
• ...

この要領で、SlidingTopK Redisを更新する。(write-through-update方式)

ただ、この方法だと、8:01　に出てきたバズ動画が、9:00になるまでランキングに入らない。
この解決策として、最新1分の top500を取得し、現在 Redis Sliding TopK に入っていない動画があれば、その過去60分間のカウントをクエリし、以降集計を継続する。

🟢 Great Solution: Query Top K every minute

よく考えると、上記のやり方をしなくても、毎分、Cron Jobを過去60分間のTopKを PerMinuteCount のシャードからクエリしてマージし、Redis Sliding TopK に書き込むので良い。

DD Can we make use of approximations to improve performance? [HARD] 🔥

→ CountMin Sketch と SortedSet を組み合わせて使う。

🟢 Good Approach: Use Redis Count-Min Sketch x Redis Sorted Set

DD の Kafka, GlobalPer*Job 以降が不要になる。
以下の手順になる。

1. LocalPerMinJob は video_id ごとに、(update count-min sketch && get appoximage count) をLua script で実行する。
2. LocalPerMinJob は、SortedSet を見て、得られたカウントがTop 2Kに食い込むようであれば、Redis Sorted Set の更新を行う。

ただし、これだと、Redis Count-Min Sketch および Redis Sorted Set への書き込みが Window内で閲覧されたユニークな動画数となり、非常に高負荷である。したがって、LocalPerMinJob 内で local statistic を取り、明らかに再生数が少ない動画は、スキップして良い。

Kafka  <------ LocalPerMinJob ---->  Redis Count-Min Sketch
^^^            ^^^                   Redis Sorted Set (Top 2K)
roundrobin     1 worker per          
               1 partition.          

そして最終的に、TopK Cron Job で各シャードの top 2K の sorted list をマージし、グローバルな top K を求める。

以上の通り、個々のビデオの再生回数を全て保存しなくて良い。
TODO: 全体をなぞりながら、どの部分でパフォーマンスが向上したのかを説明できるようにする。

TODO: Durability問題。Redisが落ちた時にどうするか。

TODO: DD Is there room for a specialized database here?

NOTE: これらを使う場合、技術の内部構造を説明できる必要がある。面接練習としては、Primitive な approach の習得に時間を使った方が良い。

🟢 Good Solution: TimescaleDB (Postgres + hypertables + continuous aggregates)

TODO

🟢 Good Solution: Real-time OLAP (Druid/Pinot/ClickHouse)

TODO