すべてのデータ伝送システムのための唯一の要件はデータが失われないことです。通常、データは遅れて到着するか、再度要求される可能性がありますが、失われることはありません。

この要件を満たすために、ほとんどの分散システムは少なくとも1回の配信を保証しています。「少なくとも1回限りの配信」を提供する技術は、通常「再試行、再試行、および再試行」になります。クライアントから明確な確認を受け取るまで、メッセージが配信されるとは考えません。

しかし、ユーザーとして、 少なくとも1回限りの配信は、私が望んでいるものとはまったく異なります。メッセージを1回配信したいです。そして一度だけ。

残念ながら、 正確に1回限りの配信に近いものを実現するには、不可解な設計が必要です。アーキテクチャでは、障害の各ケースを慎重に検討する必要があります-障害が発生した後、既存の実装の一部として単純に登録することはできません。そして、それでも、メッセージが一度だけ配信されるシステムを実装することはほとんど不可能です。

過去3か月間、完全に新しい重複排除システムを開発し、できるだけ多くのさまざまな障害に直面した1回限りの配信に可能な限り近づけました。

新しいシステムは、古いシステムの100倍のメッセージを追跡できますが、信頼性の向上とコストの削減により異なります。方法は次のとおりです。

問題

ほとんどのセグメントバックエンドシステムは、再試行、メッセージの再送信、ブロッキング、および2ステージコミットで障害を適切に処理します。ただし、1つの注目すべき例外があります。 公開APIにデータを直接送信するクライアントです 。

クライアント（特にモバイルクライアント）は、データを送信できるときに通信の中断をしばしば経験しますが、APIからの応答をスキップします。

あなたがバスに乗り、iPhoneのHotelTonightアプリから部屋を予約したと想像してください。アプリケーションはセグメントサーバーへのデータのアップロードを開始しますが、バスが突然トンネルに入り、接続が失われます。送信したイベントの一部は既に処理されていますが、クライアントはサーバーからの応答を受信しません。

このような場合、サーバーが以前にまったく同じメッセージを技術的に既に受信しているにもかかわらず、クライアントは同じイベントをセグメントAPIに繰り返し送信します。

サーバーの統計から判断すると、過去4週間に受信したイベントの約0.6％は、既に受信したメッセージの繰り返しです。

エラーのレベルは取るに足らないように見えるかもしれません。しかし、数十億ドルの収益を生み出すeコマースアプリケーションの場合、 0.6％の差は、数百万ドルの利益と損失の差を意味する場合があります。

メッセージ重複排除

したがって、問題の本質を理解しています。APIに送信される重複メッセージを削除する必要があります。しかし、それを行う方法は？

理論レベルでは、重複排除システムの高レベルAPIは単純に見えます。 Python（ 別名pseudo-pseudo-code ）では、次のように表現できます。

def dedupe(stream): for message in stream: if has_seen(message.id): discard(message) else: publish_and_commit(message)

ストリーム内の各メッセージについて、そのメッセージが以前に（一意の識別子によって）検出されたかどうかが最初にチェックされます。会ったら、それを捨てなさい。会っていない場合は、メッセージを再リリースしてアトミックに転送します。

すべてのメッセージを永続的に保存しないために、有効期限までのキーの保存時間として定義される「重複排除ウィンドウ」が動作します。メッセージがウィンドウに収まらない場合、それらは廃止と見なされます。このIDを持つ1つのメッセージのみがウィンドウで送信されるようにします。

この動作は簡単に説明できますが、特別な注意が必要な詳細が2つあります。 読み取り/書き込みパフォーマンスと精度です。

システムは、データストリーム内の数十億のイベントを重複排除し、同時に低遅延で費用対効果の高い方法で重複排除を行います。

さらに、登録されたイベントに関する情報が確実に保存され、障害が発生した場合に復元できるようにし、ディスプレイにメッセージが繰り返し表示されないようにする必要があります。

建築

これを実現するために、Kafkaからデータを読み取り、4週間のウィンドウに既に記録されている重複イベントを削除する「2段階」アーキテクチャを作成しました。

高レベル重複排除アーキテクチャ

カフカトポロジー

このアーキテクチャがどのように機能するかを理解するには、まずKafkaフロートポロジを見てください。すべての着信API呼び出しは個別のメッセージに分割され、Kafka入力セクションを明確に表します。

最初に、各着信メッセージは、クライアント側で生成される一意のmessageId

でマークされます。これは通常UUIDv4です（ただし、ksuidへの切り替えを検討しています）。クライアントがmessageIdを報告しない場合、APIレベルで自動的に割り当てます。

クライアント側を複雑にしたくないため、ベクタークロックやシリアル番号は使用しません。 UUIDを使用すると、ほぼすべての主要なプログラミング言語がUUIDをサポートするため、 誰でも簡単にAPIにデータを送信できます。

 { "messageId": "ajs-65707fcf61352427e8f1666f0e7f6090", "anonymousId": "e7bd0e18-57e9-4ef4-928a-4ccc0b189d18", "timestamp": "2017-06-26T14:38:23.264Z", "type": "page" }

耐久性と再現性のために、個別のメッセージがKafkaジャーナルに記録されます。それらはmessageIdによって配信されるため、同じmessageId

常に同じハンドラーに到着することを確認できます。

これは、データ処理に関して重要な詳細です。 数千億のメッセージの中からキーの中央データベースを検索する代わりに、検索クエリを特定のセクションにリダイレクトするだけで、検索スペースを桁違いに狭めることができました。

重複排除ワーカーは、Kafka入力セクションを読み取るGoプログラムです。彼女は、メッセージの読み取り、重複のチェック、およびメッセージが新しい場合はKafka出力トピックへの送信を担当します。

私たちの経験では、Kafkaワーカーとトポロジは管理が非常に簡単です。フェールオーバーレプリカを必要とする大規模なMemcachedインスタンスはもうありません。代わりに、組み込みのRocksDBデータベースを使用しました。これは調整をまったく必要とせず、非常に低価格で永続的なストレージを提供します。これについて詳しく説明します。

Worker RocksDB

各ワーカーは、ローカルのRocksDBデータベースをローカルのEBSハードドライブに保存します。 RocksDBは、 Facebookによって開発された統合されたキーと値のリポジトリであり、非常に高いパフォーマンスのために最適化されています。

入力パーティションからイベントが取得されるたびに、コンシューマはRocksDBにそのようなmessageId

以前に検出messageId

かどうかを確認するよう要求します。

メッセージがRocksDBにない場合、キーをデータベースに追加してから、Kafkaの出力セクションでメッセージを公開します。

メッセージが既にRocksDBにある場合、ワーカーは単にメッセージを出力セクションに公開せず、メッセージを処理したという通知で入力セクションを更新します。

性能

データベースで高いパフォーマンスを実現するには、処理されるイベントごとに3種類のクエリに対応する必要があります。

入力に到着するが、データベースに保存されそうにないランダムなキーの存在を検出します。キースペースのどこにでも配置できます。
高性能で新しいキーを記録します。
「重複排除ウィンドウ」に該当しない古いキーを宣言します。

その結果、データベース全体を継続的にスキャンし、新しいキーを追加し、古いキーを廃止する必要があります。そして理想的には、これは古いデータモデルのフレームワーク内で行われるべきです。

私たちのデータベースは、非常に異なる3つのタイプのクエリを満たす必要があります。

一般的に、パフォーマンスの向上の大部分はデータベースのパフォーマンスによるものです。したがって、RocksDBデバイスを把握することは理にかなっています。

RocksDBはログ構造ツリー（LSMツリー）です。つまり、ディスク上の先行書き込みログに新しいキーを継続的に追加し、 memtableの一部としてソートされたキーをメモリに保存します。

キーは、memtableの一部としてメモリ内でソートされます

キーの作成は非常に高速なプロセスです。新しいアイテムはログに追加することでディスクに直接書き込まれ（障害が発生した場合の直接保存と回復のため）、データレコードはメモリ内でソートされ、迅速な検索と部分記録が提供されます。

十分な数のエントリがmemtableに到着するたびに、 SSTable （ソートされた行のテーブル）としてディスクに保存されます。行はすでにメモリ内でソートされているため、直接ディスクにフラッシュできます。

memtableの現在の状態は、レベル0（レベル0）のSSTableとしてディスクにフラッシュされます

作業ログからのこのようなリセットの例を次に示します。

[JOB 40] Syncing log #655020 
      

        
        
        
      

     [default] [JOB 40] Flushing memtable with next log file: 655022 
      

        
        
        
      

     [default] [JOB 40] Level-0 flush table #655023: started 
      

        
        
        
      

     [default] [JOB 40] Level-0 flush table #655023: 15153564 bytes OK 
      

        
        
        
      

     [JOB 40] Try to delete WAL files size 12238598, prev total WAL file size 24346413, number of live WAL files 3.

各SSTableテーブルは変更されないままです-作成後、変更されることはありません-これにより、新しいキーの書き込みは非常に迅速に行われます。ファイルを更新する必要はなく、レコードは新しいレコードを生成しません。代わりに、同じ「レベル」にある複数のSSTableテーブルは、帯域外圧縮フェーズ中に単一のファイルにマージされます。

個々のSSTableテーブルが1つのレベルから圧縮されると、それらのキーがマージされ、新しいファイルがより高いレベルに転送されます。このようなシールの例は、作業ログに記載されています。この場合、プロセス41は4つのゼロレベルファイルを圧縮し、それらを結合してより大きな第1レベルファイルにします。

/data/dedupe.db$ head -1000 LOG | grep "JOB 41" 
      

        
        
        
      

     [JOB 41] Compacting 4@0 + 4@1 files to L1, score 1.00 
      

        
        
        
      

     [default] [JOB 41] Generated table #655024: 1550991 keys, 69310820 bytes 
      

        
        
        
      

     [default] [JOB 41] Generated table #655025: 1556181 keys, 69315779 bytes 
      

        
        
        
      

     [default] [JOB 41] Generated table #655026: 797409 keys, 35651472 bytes 
      

        
        
        
      

     [default] [JOB 41] Generated table #655027: 1612608 keys, 69391908 bytes 
      

        
        
        
      

     [default] [JOB 41] Generated table #655028: 462217 keys, 19957191 bytes 
      

        
        
        
      

     [default] [JOB 41] Compacted 4@0 + 4@1 files to L1 => 263627170 bytes

要約が完了すると、結合されたSSTableテーブルはデータベースレコードの最終セットになり、古いSSTableテーブルはチェーン解除されます。

作業インスタンスを見ると、この主要な書き込みログがどのように更新されるか、個々のSSTableテーブルがどのように書き込まれ、読み取られ、マージされるかがわかります。

ジャーナルおよび最新のSSTableテーブルは、I / O操作の大部分を占めています

本番サーバーでSSTable統計を見ると、4つの「レベル」のファイルがあり、各レベルでファイルサイズが大きくなっています。

 ** Compaction Stats [default] ** Level Files Size(MB} Score Read(GB} Rn(GB} Rnp1(GB} Write(GB} Wnew(GB} Moved(GB} W-Amp -------------------------------------------------------------------------------------------- L0 1/0 14.46 0.2 0.0 0.0 0.0 0.1 0.1 0.0 0.0 L1 4/0 194.95 0.8 0.5 0.1 0.4 0.5 0.1 0.0 4.7 L2 48/0 2551.71 1.0 1.4 0.1 1.3 1.4 0.1 0.0 10.7 L3 351/0 21735.77 0.8 2.0 0.1 1.9 1.9 -0.0 0.0 14.3 Sum 404/0 24496.89 0.0 3.9 0.4 3.5 3.9 0.3 0.0 34.2 Int 0/0 0.00 0.0 3.9 0.4 3.5 3.9 0.3 0.0 34.2

 Rd(MB/s} Wr(MB/s} Comp(sec} Comp(cnt} Avg(sec} KeyIn KeyDrop 0.0 15.6 7 8 0.925 0 0 20.9 20.8 26 2 12.764 12M 40 19.4 19.4 73 2 36.524 34M 14 18.1 16.9 112 2 56.138 52M 3378K 18.2 18.1 218 14 15.589 98M 3378K 18.2 18.1 218 14 15.589 98M 3378K

RocksDB は、特定のSSTableテーブルのインデックスとブルームフィルターをこれらのテーブル自体に保存し、メモリにロードします。これらのフィルターとインデックスは、特定のキーを見つけるためにポーリングされ、完全なSSTableテーブルがLRUの一部としてメモリにロードされます。

ほとんどの場合、古典的な重複排除システムをブルームフィルターを使用する古典的なケースにする新しいメッセージが表示されます。

ブルームフィルターは、キーが「おそらく複数形」か「間違いなく複数形か」を示します。答えを出すために、フィルターは、以前に発生した各要素に異なるハッシュ関数を適用した後、多くのビットを節約します。ハッシュ関数のすべてのビットがセットに収束する場合、「おそらくセットに属している」という答えが返されます。

セットに{x、y、z}のみが含まれる場合、ブルームフィルターの文字wを照会します。ビットの1つが収束しないため、フィルターは「セットに属していません」という答えを返します。

答えが「おそらくセットに属する」場合、RocksDBはSSTableテーブルからソースデータを要求し、要素が実際にセットに存在するかどうかを判断できます。しかし、ほとんどの場合、フィルターは「間違いなくセットに属していません」という答えを返すため、一般にテーブルのクエリを回避できます。

RocksDBに目を向けると、要求する関連するすべてのmessageId

に対してMultiGet要求を作成します。パフォーマンスと多数の同時ブロック操作を回避するために、パッケージの一部として作成します。また、Kafkaからのデータをパッケージ化することができ、通常は連続書き込みを優先してランダム書き込みを回避します。

これは、読み取り/書き込みタスクがどのように高いパフォーマンスを発揮するかを説明しますが、古いデータがどのように陳腐化したとみなされるかという疑問が残っています。

削除：時間ではなくサイズにスナップ

重複排除プロセスでは、システムを厳密な「重複排除ウィンドウ」に制限するか、ディスク上のデータベースの合計サイズに制限するかを決定する必要があります。

すべてのユーザーの重複排除によるシステムクラッシュを回避するために、 時間間隔の制限ではなくサイズの 制限を選択することにしました。これにより、各RocksDBインスタンスの最大サイズを設定し、突然のジャンプや負荷の増加に対処できます。副作用は、時間間隔を24時間未満に短縮できることです。この境界では、当直のエンジニアが呼び出されます。

RocksDBの古いキーは、サイズが無制限に大きくなるのを防ぐために、定期的に廃止と宣言します。これを行うために、最も古いキーを最初に削除できるように、 セカンダリキーインデックスをシリアル番号で保存します。

データベースを開くときに固定TTLを維持する必要があるRocksDB TTLを使用する代わりに、ネストされた各キーのシリアル番号でオブジェクト自体を削除します。

シリアル番号はセカンダリインデックスとして保存されるため、シリアル番号をすばやく要求し、削除済みとして「マーク」することができます。シリアル番号を渡した後の削除関数は次のとおりです。

 func (d *DB) delete(n int) error { // open a connection to RocksDB ro := rocksdb.NewDefaultReadOptions() defer ro.Destroy() // find our offset to seek through for writing deletes hint, err := d.GetBytes(ro, []byte("seek_hint")) if err != nil { return err } it := d.NewIteratorCF(ro, d.seq) defer it.Close() // seek to the first key, this is a small // optimization to ensure we don't use `.SeekToFirst()` // since it has to skip through a lot of tombstones. if len(hint) > 0 { it.Seek(hint) } else { it.SeekToFirst() } seqs := make([][]byte, 0, n) keys := make([][]byte, 0, n) // look through our sequence numbers, counting up // append any data keys that we find to our set to be // deleted for it.Valid() && len(seqs) < n { k, v := it.Key(), it.Value() key := make([]byte, len(k.Data())) val := make([]byte, len(v.Data())) copy(key, k.Data()) copy(val, v.Data()) seqs = append(seqs, key) keys = append(keys, val) it.Next() k.Free() v.Free() } wb := rocksdb.NewWriteBatch() wo := rocksdb.NewDefaultWriteOptions() defer wb.Destroy() defer wo.Destroy() // preserve next sequence to be deleted. // this is an optimization so we can use `.Seek()` // instead of letting `.SeekToFirst()` skip through lots of tombstones. if len(seqs) > 0 { hint, err := strconv.ParseUint(string(seqs[len(seqs)-1]), 10, 64) if err != nil { return err } buf := []byte(strconv.FormatUint(hint+1, 10)) wb.Put([]byte("seek_hint"), buf) } // we not only purge the keys, but the sequence numbers as well for i := range seqs { wb.DeleteCF(d.seq, seqs[i]) wb.Delete(keys[i]) } // finally, we persist the deletions to our database err = d.Write(wo, wb) if err != nil { return err } return it.Err() }

さらに高い書き込み速度を保証するために、RocksDBはすぐに返らず、キーを削除しません（SSTableテーブルは不変です！）。代わりに、RocksDBは「墓石」をキーに追加します。これは、ベースの圧縮プロセス中に削除されます。したがって、シーケンシャル書き込み操作中にレコードを迅速に廃止し、古いアイテムを削除する際のメモリの詰まりを回避できます。

データの正確性

数十億のメッセージの速度、スケーリング、安価な検索がどのように提供されるかについてはすでに説明しました。最後の断片が残っています-さまざまな障害が発生した場合にデータの正確性を確保する方法。

EBSスナップショットとアプリケーション

プログラマーのエラーまたはEBSの誤動作による損傷からRocksDBインスタンスを保護するために、各ハードドライブのスナップショットを定期的に取得します。 EBSはそれ自体で複製されますが、この方法は何らかの内部メカニズムによって引き起こされる損傷から保護します。特定のインスタンスが必要な場合、クライアントを一時停止できます。この時点で、対応するEBSディスクがマウント解除され、新しいインスタンスに再接続されます。パーティションIDを変更しない限り、ドライブの再接続は、データが正しいことを保証する完全に簡単な手順のままです。

ワーカーに障害が発生した場合、メッセージを失わないように、RocksDBに統合された先行書き込みログに依存しています。 RocksDBがメッセージをログに確実に保存することが保証されるまで、入力セクションからのメッセージは許可されません。

出力セクションの読み取り

この時点まで、メッセージが厳密に一度だけ配信されることを保証できる「アトミックな」ステップがなかったことにお気づきかもしれません。いつでも、ワーカーがクラッシュする可能性があります：RocksDBへの書き込み時、出力セクションへの発行時、または受信メッセージの確認時。

これらの個別システムすべてのトランザクションを一意にカバーするアトミックな「固定」ポイントが必要です。データには何らかの「真実の源」が必要です。

これは、出力セクションからの読み取りが機能する場所です。

何らかの理由でワーカーがクラッシュしたり、Kafkaでエラーが発生して再起動した場合、最初のステップは、イベントが発生したかどうかを「真実のソース」で確認することです。このソースは出力セクションです。

出力セクションでメッセージが見つかったが、RocksDBでは見つからなかった場合（およびその逆）、重複排除ワーカーは必要な編集を行ってデータベースとRocksDBを同期します。基本的に、出力パーティションを先行書き込みログと究極の真実のソースと同時に使用し、RocksDBがそれをキャプチャして検証します。

実際の仕事で

現在、当社の重複排除システムは実稼働で3か月稼働しており、その結果に非常に満足しています。数字の場合、次のようになります。

RocksDBのディスクに保存された1.5 TBのキー
古いキーを廃止する前の4週間の重複排除ウィンドウ
RocksDBインスタンスに保存された約600億のキー
2,000億のメッセージが重複排除システムを通過する

システム全体は、高速で効率的で耐障害性がありますが、非常にシンプルなアーキテクチャです。

特に、システムの2番目のバージョンには、古い重複排除システムに比べて多くの利点があります。

以前は、すべてのキーをMemcachedに保存し、アトミック演算子を使用してCAS（check-and-set）レコードの値を確認および設定し、存在しないキーを設定しました。 Memcachedはキーの公開における固定点および「原子性」として機能しました。

このようなスキームは非常にうまく機能しましたが、すべてのキーを収めるために大量のメモリが必要でした。さらに、ランダムなMemcachedの失敗を受け入れるか、メモリが大量に必要なフェイルセーフコピーを作成するコストを2倍にするかを選択する必要がありました。

Kafka / RocksDBスキームは、古いシステムのほぼすべての利点を提供しますが、信頼性が向上しています。要約すると、主な成果は次のとおりです。

ディスクへのデータの保存：キーのセット全体または完全なインデックス作成をメモリに保存するのは、容認できないほど高価です。より多くのデータをディスクに転送し、さまざまなレベルのファイルとインデックスを使用することで、コストを大幅に削減することができました。障害発生時に「コールド」ストレージ（EBS）に切り替えることができ、障害発生時に追加の「ホット」インスタンスの動作をサポートしません。

パーティション分割：もちろん、検索スペースを絞り込み、メモリに大量のインデックスをロードしないようにするために、特定のメッセージが正しいワーカーに送信されることが保証されている必要があります。 Kafkaのパーティショニングにより、これらのメッセージを常に正しいルートにルーティングできるため、データをキャッシュし、リクエストをより効率的に生成できます。

廃止されたキーの正確な認識 ：Memcachedでは、各キーのTTLを設定してその有効期間を決定し、Memcachedプロセスに依存してキーを除外します。大きなデータパケットの場合、これはメモリの不足を脅かし、多数の主要な例外のためにCPU使用率が急上昇します。キーを削除するようにクライアントに指示することにより、「重複排除ウィンドウ」を削減することで問題を回避できます。

真実の源としてのカフカ ：複数の注視点による重複排除を本当に回避するために、下流のすべての顧客に共通する真実の源を使用する必要があります。そのような「真実の源」としてのカフカは驚くほどうまく機能します。ほとんどの障害の場合（Kafka自体の障害を除く）、メッセージはKafkaに書き込まれるか、記録されません。また、Kafkaを使用すると、大量のデータをメモリに保存することなく、公開されたメッセージが適切に配信され、複数のマシンのディスク間で複製されます。

バッチ読み取りおよび書き込み： KafkaおよびRocksDBの呼び出しに対してバッチI / O操作を行うことにより、シーケンシャル読み取りおよび書き込みを使用してパフォーマンスを大幅に改善することができました。 Memcachedを使用した以前のランダムアクセスではなく、ディスクのパフォーマンスを向上させ、インデックスのみをメモリに保存することで、スループットを大幅に向上させました。

一般に、作成した重複排除システムが提供する保証に非常に満足しています。KafkaとRocksDBをストリーミングアプリケーションの基盤として使用することがますます標準になっています。そして、この基盤の上で新しい分散アプリケーションの開発を喜んで続けます。

何十億ものメッセージを厳密に1回配信する

問題