Spotify：イベントサブシステムをGoogle Cloudに移行する（パート1）

ユーザーがSpotifyクライアントでアクションを実行すると（たとえば、曲を聞いたり、アーティストを検索したりするたびに）、少量の情報（イベント）がサーバーに送信されます。 イベント配信は、世界中のお客様から当社の中央処理システムに情報を安全かつ確実に輸送するプロセスであり、興味深いタスクです。 これらの一連の記事では、この分野で実装されているソリューションのいくつかを検討します。 より正確には、新しいイベント配信システムのアーキテクチャを見て、Google Cloudにデプロイすることにした理由を説明します。



この最初の記事では、現在のイベント配信システムがどのように機能するかを説明し、それを使用して学んだいくつかの教訓について説明します。 次に 、新しいシステムの作成と、すべてのイベントのトランスポートメカニズムとしてCloud Pub / Subを選択した理由を検討します。 3番目の最後の記事では、DataFlowを使用してすべてのイベントを処理する方法と、それがどれほど迅速に発生するかについて説明します。







配信システムを通じて配信されるイベントには多くの用途があります。 製品設計における当社のソリューションのほとんどは、A / Bテストの結果に基づいており、その結果、大規模で正確なデータに依存する必要があります。 2015年にリリースされたDiscover Weeklyプレイリストは、すぐにSpotifyの最もよく使用される機能の1つになりました。 音楽再生データに基づいて作成されます。 音楽の年 、 Spotify Party 、その他の多くのSpotify機能もデータベースに基づいています。 さらに、Spotifyデータは、 ビルボードトップをコンパイルするためのソースの1つです。



メッセージングシステムは、Spotifyデータインフラストラクチャの基本的な部分の1つです。 そのための重要な要件は、十分に説明されたインターフェイスを介して、開発者が予測可能な遅延と可用性ですべてのデータを配信することです。 使用状況データは、ある時点で事前に設定されたアクションへの応答として形成された構造化イベントのセットとして説明できます。



Spotifyが使用するイベントのほとんどは、特定のユーザーアクションに対する応答としてSpotifyクライアントによって直接生成されます。 Spotifyクライアントでイベントが発生するたびに、Spotifyゲートウェイの1つにイベントに関する情報が送信され、Spotifyゲートウェイはそのイベントをシステムログに書き込みます。 そこで、メッセージ配信システムで使用されるタイムスタンプが割り当てられます。 メッセージ配信の特定の遅延と完全性を保証するために、サーバーに到着する前にイベントを制御できないため、クライアントではなくイベントにログラベル（syslogタイムスタンプ）を使用することが決定されました。



Spotifyの場合、すべてのデータを中央のHadoopクラスターに配信する必要があります。 データを収集するSpotifyサーバーは、2つの大陸のいくつかのデータセンターにあります。 データセンター間の帯域幅は希少なリソースであり、データ伝送を特別な注意で扱う必要があります。



データインターフェイスは、Hadoop内のデータの場所と保存される形式によって決まります。 サービスによって配信されるすべてのデータは、HDFSでAvro形式で記録されます。 配信されたデータは、60分間（時間）セクション（パーティション）に分割されます。 これは、最初のメッセージ配信システムがscpコマンドに基づいており、Hadoop上のすべてのサーバーからsyslogファイルを1時間ごとにコピーしていた過去の遺物です。 現在、Spotifyのすべてのデータ処理プロセスは1時間ごとのデータに基づいているため、このインターフェースは近い将来に残るでしょう。



Spotifyのほとんどのデータプロセスは、監視アセンブリから1回だけデータを読み取ります。 一部のプロセスの出力値は、他のプロセスへの入力として機能するため、変換の長いチェーンを形成します。 プロセスが1時間データを処理した後、この元の時間に変更のチェックを実行しなくなります。 データが変更された場合、これらの変更をさらに再現する唯一の方法は、この特定の間隔（時間）ですべての関連タスク（および関連タスク）を手動で再起動することです。 これは高価で時間のかかるプロセスです。そのため、メッセージ配信サービスにこのような要件を提示し、時間セットを提供した後、そのデータを補足できなくなりました。 データの完全性の問題として知られるこの問題は、データ処理の遅延を最小限に抑えるという要件に反しています。 データの完全性の問題に関する興味深い視点は、GoogleのDataflowレポートに示されています。

オリジナルのメッセージ配信システム

システム構成

最初のメッセージングシステムは、Kafka 0.7の上に構築されました。







その中で、イベント配信システムは、1時間ごとのファイルの抽象化を中心に構築されています。 サービスマシンからのイベントを含むログファイルをHDFSにストリーミングするように設計されています。 特定の時間内にすべてのログファイルがHDFSに転送された後、タブ付きのテキストからAvro形式に変換されます。



システムが最初に作成されたとき、Kafka 0.7に欠けていた機能の1つは、Kafka Brokerクラスターが信頼性の高い永続ストレージを操作できることでした。 これは、データプロデューサー、Kafka Syslogプロデューサー、およびHadoopの間で一定の状態を維持しないという重要な設計決定に影響を与えました。 イベントは、HDFS上のファイルに書き込まれたときにのみ安全に保存されると見なされます。



Hadoop内でのみイベントが確実に存在する問題は、Hadoopクラスターがメッセージ配信システムの単一障害点になることです。 Hadoopが失敗すると、配信システム全体が停止します。 これに対処するには、イベントを収集するすべてのサービスに十分なディスク容量があることを確認する必要があります。 Hadoopがサービスに復帰したら、その状態に「追いつく」必要があり、すべてのデータをできるだけ早く転送します。 復旧時間は、主にデータセンター間で使用できる帯域幅によって制限されます。



プロデューサーは、Hadoopにイベントを送信するすべてのホストで実行されるデーモンです。 ログファイルを追跡し、ログパッケージをKafka Syslog Consumerに送信します。 プロデューサーは、イベントのタイプやイベントのプロパティについて何も知りません。 彼の観点から見ると、イベントはファイル内の一連の行であり、すべての行は同じチャネルにリダイレクトされます。 つまり、1つのログファイルに含まれるすべてのタイプのイベントも1つのチャネルを介して送信されます。 このようなシステムでは、Kafkaトピックはイベントを送信するためのチャネルとして使用されます。 プロデューサーがログをコンシューマーに送信した後、コンシューマーがログ行をHDFSに正常に保存したことの確認（ACK）を待つ必要があります。 プロデューサーは、送信されたログのACKを受信した後にのみ、それらが安全に保存されていると考え、他のレコードを転送します。



イベントの場合、プロデューサーからコンシューマーに到達するには、Kafka Brokersを通過してから、Kafka Groupersを通過する必要があります。 Kafka BrokersはKafkaの標準コンポーネントであり、Kafka Groupersは私たちが書いたコンポーネントです。 グルーパーは、ローカルデータセンターからのすべてのイベントストリームを処理し、それらを再び圧縮して公開し、1つのトピックに効果的にグループ化し、コンシューマーがプルします。



抽出、変換、ロード（ETL）タスクは、タブで区切られた単純な形式のデータをAvro形式に変換するために使用されます。 このプロセスは通常のHadoop MapReduceジョブであり、 Crunchフレームワークを使用して実装され、1時間ごとのセットで動作します。 特定の時間に作業を開始する前に、彼はすべてのファイルが完全に転送されることを確認する必要があります。



すべてのプロデューサーは、ファイルの終わりトークンを含む可能性のあるチェックマークを常に送信しています。 Producerは、ファイル全体が安全にHadoopに保存されているとProducerが判断したときに1回だけ送信されます。 状態（または「サバイバビリティ」）モニターは、特定の時間にサービスマシンが実行されていたすべてのデータセンターのサービス検出システムを常にポーリングします。 この1時間ですべてのファイルが最終的に転送されたかどうかを確認するために、ETLは、データの終わりが予想されるサーバーに関する情報をファイルの終わりマーカーと比較します。 ETLが不一致と不完全なデータ転送を判断した場合、ETLはデータの処理を特定の時間遅延させます。



一般的なHadoop MapReduceタスクである既存のマッパーおよびレデューサーであるETLを最大限に活用できるようにするには、入力データをシャーディングする方法を知る必要があります。 マッパーとレデューサーは、入力データのサイズに基づいて計算されます。 最適なシャーディングは、コンシューマーから継続的に受信されるイベントの数に基づいて計算されます。

レッスン

この設計に関連する主な問題の1つは、ローカルのプロデューサーがデータが確実に配信されると見なされる前に、中央の場所のHDFSに保存されることを確認する必要があることです。 これは、米国西海岸のProducerサーバーが、ロンドンのディスクにデータが書き込まれていることを知る必要があることを意味します。 ほとんどの場合、正常に機能しますが、データ転送が遅くなると、配信の遅延が発生し、それを取り除くのが困難になります。



サービスポイントがローカルデータセンターにある場合のオプションと比較してください。 通常、データセンター内のホスト間のネットワークは非常に信頼性が高いため、これによりプロデューサーの設計が簡素化されます。



問題を要約すると、世界中の1秒あたり700,000を超えるイベントを確実に配信できるシステムに非常に満足しています。 システムの再設計により、ソフトウェア開発プロセスを改善する機会も与えられました。



1つのチャネルを介してすべてのイベントを一緒に送信することにより、異なるQuality of Service（QoS）でイベントフローを管理する柔軟性が失われました。 また、リアルタイムで動作するプロセスは、ストリーム全体が入っている単一のチャネルを介してデータを転送し、そこから必要なもののみを除外する必要があるため、リアルタイムでの作業も制限されていました。



非構造化データの転送では、追加のETL変換が必要になるため、不要な遅延が追加されます。 現在、ETL作業により、イベント配信に約30分の遅延が追加されます。 データがAvro形式で送信された場合、HDFSでの記録時にすぐに利用できました。



送信者が時間の終わりを追跡する必要も問題を引き起こしました。 たとえば、マシンが停止した場合、ファイルの終わりに関するメッセージを送信できません。 ファイルの終わりマーカーが失われた場合、このプロセスが手動で中断されるまで永遠に待機します。 車の数が増えると、この問題はますます緊急になります。

次のステップ

Spotifyで配信されるメッセージの数は常に増加しています。 負荷が増加した結果、より多くの問題が発生し始めました。 時間が経つにつれて、停止の数は私たちを心配し始めました。 私たちもシステムも、増加した負荷に対処できないことに気付きました。 次の記事で 、システムの変更を決定した方法について説明します。



特定の時点でシステムによって処理されたメッセージの数。