Logbroker：Yandexでの大量のデータの収集と配信

こんにちは 私の名前はアレックス・オゼリツキーです。 Yandexでは、テクノロジーとインフラストラクチャの開発に取り組んでいます。 何百万人もの人々が使用するサービスだけでなく、非常に大量のデータを障害なく操作できることが重要です。 当社の主要な内部ツールの1つはYです。統計は、Yandexの従業員専用であり、さらに企業秘密です。 統計は、Yandexサービスから情報（主にログ）を収集、保存、処理します。 彼女との仕事の結果は、さらなる分析と製品の意思決定のための統計計算です。

統計の重要なコンポーネントの1つは、分散型のマルチセンターデータ収集および配信ソリューションであるLogbrokerです。 システムの主要な機能は、データセンターの切断に耐える機能、メッセージ配信のセマンティクスの1回のサポート、リアルタイムストリーム（ソースでのイベントの発生からレシーバーでの受信までの遅延の秒）のサポートです。

システムの中核はApache Kafkaです。 LogbrokerはAPIを使用して、未加工のApache Kafkaストリームからユーザーを分離し、災害復旧プロセス（1回限りのセマンティクスを含む）およびサービスプロセス（センター間レプリケーション、計算クラスターへのデータの配信：YT、YaMR ...）を実装します。

物語

最初、Ya.StatistikiはOracleを使用してレポートを計算しました。 データはサーバーからダウンロードされ、Oracleに書き込まれました。 1日1回、SQLクエリを使用して、データに関するレポートが作成されました。

データ量は毎年2.5倍に増加したため、2010年までにそのようなスキームは失敗し始めました。 データベースは負荷に対処するのをやめ、レポートの計算は遅くなりました。 データベースからのレポートの計算はYaMRに転送され、ログの長期保存として、各自のデータが個別のファイルとして機能する分散ストレージである自己記述ソリューションの使用が開始されました。 ソリューションは、単にアーカイブと呼ばれていました。

アーカイブには、現在のLogbrokerの多くの機能がありました。 特に、彼はデータを受信し、データをさまざまなYaMRクラスターと外部コンシューマーに配布できました。 アーカイブはメタデータも保存し、さまざまなパラメーター（時間、ファイル名、サーバー名、ログタイプなど）で選択を取得できるようにしました。 YaMRでデータの障害または損失が発生した場合、アーカイブから任意のデータを復元することができました。

サーバーをバイパスし、データをダウンロードしてアーカイブに保存しました。 2011年初頭、このアプローチは滑り始めました。 明らかなセキュリティ問題（統計サーバーはすべてのYandexサーバーにアクセスできる）に加えて、彼はスケーラビリティとパフォーマンスの問題を抱えていました。 その結果、サーバーにインストールされ、ログを直接アーカイブにアップロードするクライアントが作成されました。 クライアント（内部製品名push-client）は、外部の依存関係を使用せず、Linux、FreeBSD、さらにはWindowsのさまざまなバージョンをサポートする軽量のCプログラムです。 このプロセスでは、クライアントはデータファイル（ログ）の更新を監視し、http（s）プロトコルを使用して新しいデータを送信します。

アーカイブにはいくつかの欠点がありました。 彼はデータセンターの切断について心配しませんでした。なぜなら、彼は拡張性に制限のあるファイル編成であり、基本的にリアルタイムでのデータの送受信をサポートしていなかったからです。 アーカイブにデータを送信する前に、消費者が中古配達の遅延を望んでいた間に、少なくとも1分間隔でバッファーを蓄積する必要がありました。

フォールトトレラントストレージで利用可能なNoSQLソリューションの1つを基本として、新しいアーカイブを作成することが決定されました。

HBase 、 Cassandra 、 Ellipticsを確認しました 。 HBaseは、クロスセンターレプリケーションの問題のため適合しませんでした。 CassandraとEllipticsは、データベースの最適化中およびクラスターに新しいマシンを追加する際のパフォーマンスが大幅に低下したため適合しませんでした。 原則として、これはまれな操作であるため、新しいマシンの追加の制限は依然として克服できますが、デフラグの制限は重要であることが判明しました：毎日のデータセットはアーカイブに毎日書き込まれ、古いデータはN日前に削除されます、つまりデフラグ継続的に実行する必要があります。

最終的に、配信アーキテクチャを確認することにしました。 長期保存の代わりに-メタデータのフルセットとあらゆる種類のスライスを構築する機能を備えたアーカイブでは、サンプルを構築するための最小限の機能を備えた簡単で迅速な短期保存を作成することにしました。

実装はApache Kafkaに基づいていました。

技術的な詳細

Kafkaは永続メッセージキューを実装しています。 永続性とは、Kafkaに送信されたデータがプロセスまたはマシンを再起動した後も存在し続けることを意味します。 Kafkaの同じタイプのメッセージの各ストリームは、 トピックと呼ばれるエンティティによって表されます 。 トピックはパーティションに分割されます 。これにより、トピックの並列書き込みと読み取りが保証されます。 Kafkaは非常に簡単にスケーリングできます。フローの量が増えると、トピックパーティションの数とそれらを提供するマシンの数を増やすことができます。

パーティションは複数のレプリカに存在し、さまざまなノード（Kafkaの用語ではブローカー ）によって提供されます。 レプリカには、リーダーとフォロワーの2種類があります。 リーダーレプリカはデータの書き込みと読み取りのすべての要求を受け入れ、フォロワーレプリカはリーダーレプリカからのストリームを受け入れて保存します。 Kafkaは、すべてのレプリカのバイト単位のアイデンティティを保証します。 ブローカーの1つが使用できなくなった場合、すべての主要レプリカは自動的に他のブローカーに移動し、パーティションの書き込みと読み取りは引き続き可能です。 ブローカーが発生すると、彼がサービスを提供したすべてのレプリカが自動的にフォロワーになり、レプリカリーダーから蓄積されたバックログに追いつきます。 時々、Kafkaはリーダーのバランスをとるプロセスを開始する可能性があるため、通常の状況では、一部のブローカーがリクエストで過負荷になり、一部のブローカーが過負荷になる状況はありません。



Kafkaクラスターへの書き込みはProducerというエンティティによって行われ、読み取りはコンシューマによって行われます。

トピックは、同じタイプのメッセージのキューです。 たとえば、KafkaでWebサーバーのアクセスログを作成する場合、nginxログエントリを1つのトピックに配置し、apacheログを別のトピックに配置できます。 Kafkaはトピックごとに、次のようなパーティション化されたログを保存します。




各パーティションはメッセージの順序付けられたシーケンスであり、要素の順序は変更されません。 新しいメッセージが到着すると、最後に追加されます。 キュー内の各メッセージには一意の識別子があります（Kafkaに関してオフセット）。 メッセージには順番に番号が付けられます（1、2、3、...）。

定期的に、Kafka自体が古いメッセージを削除します。 ディスク上のパーティションは物理的にファイルセグメントのセットで表されるため、これは非常に安価な操作です。 各セグメントには、シーケンスの連続部分が含まれます。 たとえば、識別子が1、2、3、4、5、6のメッセージがあるセグメントがありますが、識別子が1、3、4、2、5、6のメッセージがあるセグメントはありません。古いメッセージを削除すると、単に削除されます最も古いファイルセグメント。

メッセージ配信にはさまざまなセマンティクスがあります。最大で1回-複数のメッセージが配信されず、少なくとも1回-少なくとも1つのメッセージが配信され、1回だけ-正確に1つのメッセージが配信されます。 Kafkaは、配信のために少なくとも1回はセマンティクスを提供します。 メッセージの複製は、書き込み中と読み取り中の両方で発生する可能性があります。 問題（応答を受信する前の切断、送信のタイムアウトなど）が発生した場合の標準のKafkaプロデューサーは、メッセージを再送信します。多くの場合、重複が発生します。 メッセージの読み取り時に、データ処理後にメッセージIDの修正が行われ、2つのイベント間でエラーが発生した場合、重複する可能性があります。 しかし、読み取りの場合、リーダーはプロセスを完全に制御し、データ処理と識別子の完了を同時に記録できるため、状況はより単純です（たとえば、データベースに書き込まれた場合に1つのトランザクションでこれを行う）。

LogbrokerとKafkaのリンク

ログブローカー：

• Kafkaの内部キッチンをユーザーから隔離します。
  
  • クライアントからのデータストリームをKafkaメッセージに変換します。
  • 必要なすべてのメタ情報をメッセージに追加します（サーバー名、ファイル名、ログタイプ...）。
  • 最も適切なトピック名を選択します。
• データの書き込みおよび読み取り用のhttp APIを提供します。
• 新しいデータをYTおよびYaMRクラスターに自動的にアップロードするサービスを提供します。
• Kafkaクラスター間でのセンター間データレプリケーションのサービスを提供します。
• Kafkaを中心に構成されたインフラストラクチャを提供します。
  
  • 構成サービス（ウォッチドッグ、ディスクの起動、ディスクの再調整、セグメントの強制削除）;
  • メトリック、グラファイトとの統合（http://graphite.wikidot.com/） ;
  • 監視用のスクリプト。

当初、Logbrokerサービスは古いアーカイブと完全にAPI互換であったため、移動しても顧客は違いに気付きませんでした。

古いAPIには、次の記述方法がありました。

• /ストア（データレコード）、
• / commit_request（トランザクションコミットリクエスト）、
• /コミット（コミット）。

このAPIは、ほぼ1回限りのデータ配信のセマンティクスを提供します。 カフカの側面でどのように見えるか。

/ storeリクエストでは、一時ストレージのローカルファイルにデータを保存します。 / commit_requestリクエストに対して、このファイルをKafkaに書き込みます。ファイルにはtransaction_idが割り当てられます。 / commitリクエストに対して、特定のtransaction_idを持つファイルが書き込まれたことを伝える特別な小さなコミットメッセージをKafkaに書き込みます。

消費者は60秒のウィンドウでこのストリームをKafkaから読み取ります。 特別なコミットメッセージが見つかったtransaction_idのすべてのメッセージは、ユーザーに提供し、残りはスキップします。

クライアントの記録のタイムアウトは30秒であるため、クライアントがコミットメッセージを書き込んで送信したが、コンシューマがこのメッセージを見逃した確率はゼロです。 コミットメッセージは小さいため、そのレコードがハングする可能性はほぼゼロです。



ローンチが成功した後、厳密な1回限りのセマンティクスでリアルタイムのストリーミング配信を実現したいと考えました。

新しい（rt）プロトコルの用語：

• sourceId-ソースの一意の識別子（たとえば、特定のホスト上のファイルに対応する場合があります）;
• seqno-常に増加している一意の（sourceId内の）チャンク識別子（たとえば、ファイル内のオフセットに対応する場合があります）。

クライアントは、sourceIdごとにLogbrokerとの接続を1回確立し、小さなチャンクでデータを送信します。 Logbrokerは、Kafka sourceIdおよびseqnoのメッセージを書き込みます。 さらに、各sourceIdは常に同じKafkaパーティションに書き込まれることが保証されています。

何らかの理由で接続が切断された場合、クライアントは（おそらく別のホストで）接続を再度作成しますが、LogbrokerはこのsourceIdに関連するパーティションを読み取り、最後に記録されたseqnoを判別します。 クライアントからのチャンクにseqno <=が記録されている場合、それらはスキップされます。



パーティションを読み取るため、データ記録のためのセッションは、目立った時間（この場合は最大10秒）を作成します。 これはまれな操作であるため、配信の遅延には影響しません。 書き込み/読み取り配信サイクルを測定しました。作成から読み取りまでのチャンクの88％が1秒で収まり、99％が5秒で収まりました。

クロスセンターミラーリングの仕組み

Kafka配信のミラーは使用せず、独自に作成しました。 主な違い：

• 多数の読み取りおよび送信バッファーは作成されません。その結果、メモリの消費量が大幅に削減され、多数のパーティションで作業できます（現在の運用では最大10,000個まで可能です）。
• パーティションは1対1でミラーリングされることが保証されており、メッセージは混在していません。
• ミラーリングは、1回限りのセマンティクスを保証します。重複やメッセージの損失はありません。

物理的には、LogbrokerプロセスとKafkaプロセスは同じマシンで実行されます。 ミラーリングプロセスは、レプリカがこのマシンのリーダーであるパー​​ティションでのみ機能します。 マシンのリーダーが変更されると、ミラーリングプロセスがこれを自動的に決定し、他のパーティションのミラーリングを開始します。

ミラーリングする対象と場所を正確にどのように決定しますか？

データセンターにdc1とdc2という名前の2つのKafkaクラスターがあるとします。 データがdc1に分類されると、dc1プレフィックスを付けてトピックに書き込みます。 たとえば、dc1.log1。 dc2のデータは、dc2プレフィックスが付いたトピックに書き込まれます。 プレフィックスによって、どのトピックがこのデータセンターに最初に書き込まれたかを判断し、それらが別のデータセンターにミラーリングされます。



ミラーリングスキームには、ニュアンスが異なるオプションがある場合があります。 たとえば、Logbrokerが複数の大規模な消費者のいるデータセンターにある場合、すべてのデータセンターのすべてのトピックをここにミラーリングすることは理にかなっています。 Logbrokerが大規模なコンシューマのないデータセンターにある場合、このデータセンターのローカルデータコレクターとして機能し、大規模なLogbrokerインストールにデータを送信できます。

ミラーリングに1回だけのセマンティクスを提供するにはどうすればよいですか？

データはパーティションごとにミラーリングされます。 ミラーリング中、各パーティションに60秒ごとに2つの数値を保存します。元のパーティションのオフセット記録とミラーリングされたパーティションのオフセット記録です。 記録中に障害が発生した場合、元のパーティションとミラーリングされたパーティションのレコードに保存されている状態と実際のサイズを読み取り、これらの数値から元のパーティションの読み取り元を正確に決定します。

クロスセンターデータ配信アーキテクチャ

データソースはLogbrokerのインストールトポロジーについて何も知りません。 Logbroker自体がソースレコードを目的のノードに転送します。 ソースがLogbrokerがインストールされているデータセンターにある場合、データはそこに書き込まれ、データはミラーリングを使用して他のデータセンターに転送されます。

Logbrokerからデータを取得するには、いくつかのシナリオがあります。 まず、配信の種類（プッシュまたはプル）によって区別されます。 つまり、Logbrokerは消費者にデータを送信する（プッシュ配信）か、消費者がhttp-protocolを介してデータを読み取る（プル配信）ことができます。 どちらの場合も、2つのデータ取得シナリオをサポートしています。

消費者がLogbrokerクラスターのいずれかと同じデータセンターに完全にいる場合、すべてのトピックを読みます。

Logbrokerが表されていないデータセンターに顧客がいる場合、または顧客が複数のデータセンターに分散している場合、顧客は各データセンターからデータセンタープレフィックスが付いたトピックのみを読み取ります。 たとえば、dc1データセンターのクラスターから読み取る場合、dc1プレフィックスを持つトピックのみを読み取ります。

一見すると、1つのデータセンターからすべてのトピックを読むことができ、無効になっている場合は、別のデータセンターで同じトピックを読むように切り替えることができます。 これは実際にはそうではありません。 事実、異なるデータセンター内の異なるKafkaクラスター上の同じ名前のパーティション内のメッセージは、異なるオフセットを持つことになります。 これは、クラスターが異なる時間に起動され、ミラーリングが異なる時間に開始されたという事実によるものです。 ミラーは、メッセージの不変の順序のみを保証します。 Logbrokerのリーダーはパーティションのオフセットを操作します。別のデータセンターではパーティションが異なるため、リーダーを切り替えると、重複を受信するか、データを受信しません。

実際、多くの場合、このような読み取りスイッチは役に立ちません。 たとえば、dc1データセンターで、dc1プレフィックスが付いたトピックを読みます。 突然、dc1データセンターが崩壊しました。dc2に切り替えて、dc1プレフィックスの付いたトピックを読み続けたいと思います。 ただし、dc1データセンターが落ちたため、このプレフィックスが付いたトピックは書き込まれず、dc2データセンターには新しいデータはありません。

Logbrokerのようなエンティティが必要なのはなぜですか？ ソースからすべての消費者にすぐにデータを送信できないのはなぜですか？

Logbrokerは、データ配信スキームを大幅に簡素化します。 彼が不在の場合、彼は一度に複数の場所にデータを送信できる複雑なクライアントを作成する必要がありました。 バグを修正し、新しい消費者のサポートを追加するには、複雑なクライアントを絶えず更新する必要があります。 これで、Cで書かれた非常にシンプルなライトクライアントができました。このクライアントはめったに更新されません。 たとえば、一部のソースには、2012年の古いビルドがまだあります。これは、以前はアーカイブで動作し、現在はLogbrokerで動作します。

中央集中型配信は、センター間のトラフィックも大幅に節約します。



受信クラスターで障害が発生した場合、不足しているLogbrokerデータを迅速かつ効率的に分配できます。 ソース自体で同じ操作を行うのははるかに困難です。

現在、YandexではLogbrokerはどこで使用されていますか？

現在、Logbrokerはサーバーからログを提供するために使用されています。 着信トラフィックの合計量は、1日あたり平均60テラバイトです。 発信-3〜4倍（特定のデータセンターによって異なります）。 ログコンシューマは、主に統計クラスタと検索クラスタです。 また、必要なログのみの小さなストリームを受信する多くの小規模な消費者もいます。

現在、Logbrokerを社内のトランスポートバスとして導入し、ログだけでなくバイナリデータも送り出すように取り組んでいます。

開発プロセス中に、主にカフカの水分に関連する困難に遭遇しました。 特に、クライアントコードは非常に粗雑であることが判明しました。 データの書き込みおよび読み取り用の高レベルのライブラリは、しばらく固執し、プロセスをブロックする可能性があります。 最初はこのコードから始めましたが、開発プロセスではKafkaの非同期クライアントを作成しました。 現在、このクライアントをオープンソースライブラリとして設計することを検討しています。

クライアントとは異なり、Kafkaのサーバー側は非常に高品質で記述されており、実際に変更なしで使用されています。 私たちは時々開発者に送ろうとする小さなパッチを使用します。