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

このシリーズの最初の記事では 、古いメッセージ配信システムがどのように機能するか、およびその機能から得た結論のいくつかについて説明しました。 2番目では、新しいシステムの設計と、すべてのイベントのトランスポートメカニズムとしてCloud Pub / Subを選択した理由を検討しました。 この3番目の最後の記事では、 Dataflowを使用して公開されたすべてのイベントを処理する方法と、このアプローチについて学んだことを説明します。

Dataflowを使用して、Pub / Subから（by）時間間隔にイベントをエクスポートします

Spotifyが現在行うタスクのほとんどはバッチジョブです。 イベントを永続ストレージに確実にエクスポートする必要があります。 このような永続ストレージとして、従来はHadoop分散ファイルシステム（HDFS）とHiveを使用しています。 保存されているデータのサイズとエンジニアの数の両方で測定できるSpotifyの成長に合わせて、HDFSからクラウドストレージ 、HiveからBigQueryに徐々に切り替えています。



抽出、変換、およびロード（ETL）ジョブは、HDFSおよびクラウドストレージからデータをエクスポートするために使用するコンポーネントです。 HiveおよびBigQueryのエクスポートは、1時間ごとのアセンブリからHDFSおよびクラウドストレージにデータを変換するバッチジョブによって処理されます。



エクスポートされたすべてのデータは、タイムスタンプに従って1時間ごとのパケットに分割されます。 これは、最初のイベント配信システムで導入されたオープンインターフェイスです。 システムはscpコマンドに基づいており、1時間ごとのsyslogファイルをHDFSのすべてのサーバーからコピーしました。



ETLジョブは、監視アセンブリのすべてのデータが永続ストレージに書き込まれていることを高い確度で判断する必要があります。 ウォッチアセンブリのデータがこれ以上ない場合、完了としてマークされます。



すでに完了したアセンブリに遅れている人は、通常、実行されるタスクがアセンブリからデータを1回読み取るため、アセンブリに追加できません。 この問題を解決するには、ETLジョブで遅延データを個別に処理する必要があります。 すべての遅延データは現在の営業時間のアセンブリに記録され、イベントのタイムスタンプを未来にシフトします。



ETLジョブを作成するために、 Dataflowを試すことにしました。 この選択は、運用上の責任をできるだけ少なくしたいという事実と、他の人が大きな問題を解決するという事実によるものでした。 データフローは、データ記録をパイプライン化するためのフレームワークであり、このようなパイプラインを実行するためのGoogle Cloudの完全に管理されたサービスです。 箱から出してすぐに、Cloud Pub / Sub、Cloud Storage、BigQueryで動作します。



Dataflowでパイプラインを記述するのは、Apache Crunchでパイプラインを記述するのとよく似ています。 どちらのプロジェクトもFlumeJavaに触発されているため、これは驚くことではありません。 違いは、Dataflowにはストリーミングおよびバッチ作業用の統合モデルが用意されているのに対して、Crunchにはバッチモデルしかないことです。







エンドツーエンドの適切な遅延を実現するために、ETLをストリーミングジョブとして作成しました。 常に実行されているという事実により、データが到着すると個々の監視アセンブリを段階的に満たすことができます。 これにより、1時間ごとにデータを1回エクスポートするバッチ作業よりも遅延が少なくなります。



ELTタスクは、 ウィンドウイングデータフローの概念を使用して、時間に基づいてデータを時間アセンブリに分割します。 Dataflowでは、イベント時間と処理時間の両方でウィンドウを割り当てることができます。 タイムスタンプに基づいてウィンドウを作成できるという事実は、Dataflowが他のストリームフレームワークより優れていることを示しています。 これまでのところ、 Apache Flinkのみが時間と処理の両方でウィンドウの作業をサポートしています。



各ウィンドウは1つ以上のブロック（ペイン）で構成され、各ブロックには要素のセットが含まれます。 各ウィンドウに割り当てられるトリガーにより、ブロックの作成方法が決まります。 これらのブロックは、データがGroupByKeyを通過した後にのみ割り当てられます。 GroupByKeyはキーとウィンドウでグループ化されるため、同じブロック内のすべての集約要素は同じキーを持ち、同じウィンドウに属します。



データフローは、「ウォーターマーク」と呼ばれるメカニズムを提供します（ウォーターマークは、ここでは画像やメモとは異なり、制限や境界線を指します）。ウィンドウを閉じるタイミングを決定するために使用できます。 着信データストリームのイベント時間を使用して、特定のウィンドウのすべてのイベントがすでに到着している可能性が高い時点を計算します。

ETL実装の詳細

このセクションでは、イベント配信用のDataflow ETLタスクの作成で発生した問題のいくつかを見ていきます。 Dataflowまたは同様のシステムの経験がない場合は、理解するのが少し難しい場合があります。 （概念と用語が初めての場合）理解に役立つのは、Google DataFlowの出版物です。







イベント配信システムでは、イベントタイプとCloud Pub / Subトピックの間に1対1のマッピングがあります。 1つのETLタスクは、イベントタイプの単一ストリームで機能します。 独立したETLタスクを使用して、すべてのタイプのイベントからのデータを処理します。



使用可能なすべてのワーカーに負荷を均等に分散するために、データストリームは、各イベントにウィンドウを割り当てる変換を受ける前に分割されます。 使用するシャードの数は、タスクに割り当てられたワーカーの数の関数です。







「ウィンドウ」は複合変換です。 この変換の最初の段階では、着信ストリーム内のすべてのイベントに固定の1時間ごとのウィンドウを割り当てます。 ウォーターマークが1時間を超えると、ウィンドウは閉じられたと見なされます。

@Override public PCollection<KV<String, Iterable<Gabo.EventMessage>>> apply( final PCollection<KV<String, Gabo.EventMessage>> shardedEvents) { return shardedEvents .apply("Assign Hourly Windows", Window.<KV<String, Gabo.EventMessage>>into( FixedWindows.of(ONE_HOUR)) .withAllowedLateness(ONE_DAY) .triggering( AfterWatermark.pastEndOfWindow() .withEarlyFirings(AfterPane.elementCountAtLeast(maxEventsInFile)) .withLateFirings(AfterFirst.of(AfterPane.elementCountAtLeast(maxEventsInFile), AfterProcessingTime.pastFirstElementInPane() .plusDelayOf(TEN_SECONDS)))) .discardingFiredPanes()) .apply("Aggregate Events", GroupByKey.create()); }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

ウィンドウを割り当てるとき、ウィンドウが閉じるまでブロックごとにN個の要素を割り当てるように設定された初期トリガーがあります。 トリガーのおかげで、データが到着するとタイムパックが常に満たされます。 このセットアップトリガーは、より低いエクスポート遅延を実現するだけでなく、GroupByKeyの制限を回避するのにも役立ちます。 GroupByKeyはメモリの変換であるため、パネルで収集されるデータの量は、ワーカーのマシンのメモリに収まる必要があります。



ウィンドウが閉じると、ブロックの割り当ては後のトリガーによって制御されます。 このトリガーは、N個の要素の後、または操作時間の10秒後にデータのブロックを作成します。 イベントが1日以上遅れると、イベントは破棄されます。



ブロックの実体化（たとえば、一時ストレージまたはテーブルの作成）は、「イベント集約」変換で実行されます。これはGroupByKey変換にすぎません。





1秒あたりの着信イベントの数



ETLタスクを通過する1秒あたりの着信イベントの数を追跡するために、時間単位の割り当てウィンドウの出力で「タイムリーおよびレイトイベントの平均RPSの監視」を使用します。 すべての変換メトリックは、カスタムメトリックとしてCloud Monitoringに送信されます。 インジケーターは、毎分送信される5分間のスライドウィンドウで計算されます。



イベントの適時性に関する情報は、イベントがウィンドウに割り当てられた後にのみ取得できます。 ウィンドウ要素の最大タイムスタンプと現在の透かしを比較すると、そのような情報が得られます。 透かしデータは変換間で同期されないため、この方法での適時性の検出は不正確になる可能性があります。 現在、誤って検出されている遅延イベントの数は非常に少なく、1日1回未満です。



監視変換（または適時および遅延イベントの平均RPSの監視）が集約イベントの出力に適用される場合、イベントの適時性を正確に検出できます。 このアプローチの不利な点は、要素の数とイベントの時間に基づいてウィンドウが取得されるため、メトリックが予測できないことです。







Write to HDFS / GCS変換では、HDFSまたはCloud Storageにデータを書き込みます。 HDFSとCloud Storageでの記録の仕組みは同じです。 唯一の違いは、使用されるファイルシステムAPIです。 実装では、両方のAPIはIOChannelFactoryインターフェイスの背後に隠されています。



1つのファイルのみがブロックに書き込まれるように、障害の可能性を無視しても、各ブロックは一意のIDを受け取ります。 ブロック識別子は、記録されたすべてのファイルの一意のIDとして使用されます。 ファイルは、イベントIDスキームと一致するスキームを使用してAvro形式で書き込まれます。



タイムリーなブロックは、イベントの時間に基づいてパケット（バケット）に書き込まれます。 閉じたアセンブリの追加は、Spotifyでのデータの操作には望ましくないため、最新のものは現在の時間のパッケージに書き込まれます。 ブロックがタイムリーかどうかを理解するには、 PaneInfoオブジェクトを使用します。 ブロックが作成されるときに作成されます。



1時間ごとのアセンブリの完全性マーカーは1回だけ書き込まれます。 これを行うには、Write Paneアクションのメイン出力ストリームがクロックウィンドウに再ウィンドウ化され、Aggregated Write Successesに集約されます。





1秒あたりの記録されたファイルの数





ミリ秒の透かし遅延



メトリックの取得は、ペインの書き込みアクションの副次的な出力です。 1秒あたりに書き込まれたファイルの数、イベントの平均遅延、および現在時刻と比較した「透かし」の遅れを示すデータを取得します。 これらのメトリックはすべて、5分間のウィンドウで計算され、毎分送信されます。



HDFS / Cloud Storageに記録した後、ウォーターマークラグを測定するため、システムのレイテンシ全体に直接関連しています。 遅延のあるグラフでは、現在のキャラクターの遅延は基本的に200秒未満（約3.5分）であることがわかります。 同じ図で、最大1500秒（約25分）のランダムなバーストを確認できます。 このようなピークは、VPN経由でHadoopクラスターに記録する際の断片化が原因です。 比較のために、古いシステムのレイテンシは「最高の日」で2時間、平均で3時間です。

ETLジョブの次のステップ

ETLジョブの実装は、まだ試作段階です。 これまでに、4つのETLジョブが進行中です（1秒あたりのイベントのグラフを参照）。 最小のタスクは1秒あたり約30イベントを消費し、最大のタスクは1秒あたり10万イベントのピーク値に達します。



ETLタスクの最適なワーカー数を計算する良い方法はまだ見つかりません。 これまでのところ、それらの数は試行錯誤の後に手動で決定されます。 最小のタスクには2人のワーカーを、最大のタスクには42人のワーカーを使用します。 タスクの実行もメモリに依存していることに注意してください。 1秒あたり約20Kイベントを処理する1つのパイプラインでは24人のワーカーを使用し、2つ目では同じ速度でイベントを処理しますが、平均メッセージサイズは4倍小さく、4つだけを使用します。 自動スケーリングの機能を実装します 。



タスクを再開するときに、データが失われないようにする必要があります。 ジョブの更新が機能しない場合、これは当てはまりません。 この問題の解決策を見つけるために、Dataflowのエンジニアと積極的に協力しています。



ウォーターマークの動作はまだ謎です。 障害が発生した場合と通常の動作の場合の両方で、その計算が予測可能であることを確認する必要があります。



最後に、高速で信頼性の高いETLジョブ更新のために、適切なCI / CDモデルを定義する必要があります。 これは重要なタスクです。イベントのタイプごとに1つのETLタスクを管理する必要があり、1000を超えるタスクがあります。

クラウドイベント配信システム

私たちは、本番環境での新しいシステムの立ち上げに積極的に取り組んでいます。 実験段階での新しいシステムの立ち上げから得た予備的な数値は非常に心強いものです。 新しいシステムでの最悪の全体遅延時間は、古いプラットフォームの合計遅延時間の4分の1です。



しかし、新しいシステムから得たいのは生産性の向上だけではありません。 クラウド製品では、トランザクションコストを大幅に削減したいと考えています。 これは、Spotify製品を改善するためにより多くの時間があることを意味します。