DatabricksにおけるApache Spark™ Structured Streamingのオンデマンド状態再パーティショニングを発表

本番環境でステートフルなApache Spark™ Structured Streamingクエリを実行していると、誰もが最終的に同じ厄介な壁にぶつかることになります。

数ヶ月前にクエリを開始したとします。当時はデータ量が少なかったため、デフォルトの200個のシャッフルパーティションをそのまま受け入れて進めました。パイプラインはスムーズに動作していました。その後、ビジネスが成長してトラフィックが3倍になり、ステートストアが膨れ上がりました。突然、これら200個のパーティションは適切なサイズではなくなります。一部のパーティションに偏り（スキュー）が生じて負荷が集中し、クラスターが逼迫し、すべてのマイクロバッチの処理に想定以上の時間がかかるようになります。

そこで、当然の対応として、spark.sql.shuffle.partitionsを増やしてクエリを再起動します。しかし、何も変わりません。

最初にストリームを開始したときにパーティション数がチェックポイントに固定されてしまうため、クエリは新しい値を静かに無視します。これまでは、新しいパーティション数を適用する唯一の方法は、既存のチェックポイントを破棄して最初からやり直すことでした。しかし、ステートフルクエリにおいてこれは、慎重に維持してきた蓄積されたすべてのステートを失うことを意味します。何百万ものアカウントを追跡する不正検知モデルや、数日分のウィンドウを保持するセッショナイゼーションジョブにおいて、「最初からやり直す」という言葉は、本番環境のインシデントレビューで誰も口にしたくない言葉です。

Databricks Runtime 18以上で利用可能なオンデマンドのステート再パーティショニング（パブリックプレビュー）は、その壁を取り除きます。チェックポイントのステートを維持したまま、ステートフルなストリーミングクエリのパーティション数を変更できるようになりました。 

これは、集計、ストリーム間結合、重複排除、セッショナイゼーション、または transformWithState を実行しているかどうかにかかわらず、あらゆるステートフルなストリーミングクエリに適用され、不正検知からリアルタイム監視まで、あらゆるワークロードに対応します。

Coveoのような早期導入企業にとって、ストリーミングインフラをオンデマンドで適切なサイズに調整できる機能は、すぐに大幅な運用コストの削減につながりました。 

「Coveoでは、データ量が時間とともに大きく変動する大規模なステートフルストリーミングパイプラインを実行しています。Databricksとステート再パーティショニング機能により、関連するAmazon S3のAPIコストを40%削減できました。以前は、スケーリングを決定するたびに、オーバープロビジョニングするか、新しいチェックポイントから再構築するかというトレードオフを迫られ、ストレージAPIコストがコンピュートコストとほぼ同じになっていました。現在は、既存のステートを損なうことなく、またコストのかかるチェックポイントの移行を発生させることもなく、需要の変化に合わせて自由にスケールできます。」 —Alexis Chicoine氏（Coveo、シニアソフトウェアデベロッパー）

仕組み：なぜステートパーティションはロックされていたのか？

Coveoの成果がStructured Streamingにとってなぜ大きな飛躍を意味するのかを理解するには、そもそもなぜパーティション数が固定されていたのかを見る必要があります。

ステートフルなストリーミングクエリはステートをステートストアに保持し、そのステートは物理的にパーティショニングされています。ストリーム内の各キー（ユーザーID、アカウント番号、ウィンドウなど）は特定のパーティションにハッシュされ、各パーティションのデータはチェックポイント内の個別のRocksDBインスタンスに保存されます。パーティション数は、ディスク上のステートストア全体のレイアウトを定義します。

再起動の間にパーティション数を単純に変更すると、ハッシュが一致しなくなります。以前はあるパーティション（例えばパーティション47）に存在していたキーが、今度は別のパーティション（パーティション12）にハッシュされる可能性がありますが、その蓄積されたステートは依然として元のパーティションのファイルに残ったままになります。その結果、クエリは事実上、自身のメモリを見失うことになります。このような気付きにくいデータの破損を防ぐため、Structured Streamingはチェックポイント作成時にパーティション数をロックし、その後のspark.sql.shuffle.partitionsへの変更を無視していました。

安全ですが、柔軟性に欠けていました。その代償として、以下の2つのコストが発生していました。

1. チューニングができなかった。 200個のパーティションが不適切な選択であったと判明しても、チェックポイントの寿命が尽きるまでその設定に縛られていました。
2. ワークロードに合わせてスケールできなかった。 データ量が増減しても、パーティション数をそれに追従させることができませんでした。

オンデマンドのステート再パーティショニングは、従来の設計では拒否されていた「ステートを新しいパーティション数に合わせて物理的に再分配する」という処理を安全に行うことで、これら両方の課題を解決します。

開始するために必要なもの

要件はわずかです。

• Databricks Runtime 18以上。
• RocksDBステートストアプロバイダー。 DBR 17.3以上ではRocksDBがデフォルトであり、明示的に変更しない限り、これらのバージョンで作成された新しいクエリはRocksDBを使用します。確認または明示的に設定したい場合は、DatabricksでのRocksDBステートストアの構成を参照してください。

前提条件はこれだけです。デフォルトのステートストアを使用するDBR 18をご利用であれば、必要なものはすでにすべて揃っています。

パーティション数の変更

仕組みはシンプルで、すべてのストリーミング開発者がすでに知っている「停止、再設定、再起動」というパターンを再利用します。

spark.sql.shuffle.partitionsの代わりに、専用の設定である spark.sql.streaming.stateStore.partitionsを設定し、クエリを再起動します。

重要なポイントは、この新しい設定自体にあります。ステートフルクエリの場合、spark.sql.streaming.stateStore.partitionsはspark.sql.shuffle.partitionsよりも優先されます。これにより、従来のアプローチとは異なり、変更が確実に「反映」されます。

クエリが再起動しても、すぐに通常の処理が再開されるわけではありません。まず、保留中のものがあれば、最後に計画されたマイクロバッチを完了します。次に、1回限りの再パーティショニング処理を実行します。これは、ステートデータを新しいパーティション数に物理的に再分配し、キーを新しい正しい場所に再ハッシュすることで、データが失われたり誤って配置されたりしないようにするものです。再分配が完了すると、クエリは通常通り処理を再開し、指定したパーティション数を使用するようになります。

この再パーティショニングのステップこそが、この機能の核心です。これは、「単に数値を変更した」ことと「ステートを安全に新しいレイアウトに移行した」ことの違いを意味します。

再パーティショニング処理の監視

再パーティショニングは、実行時間がステートの量に比例する実際の処理であるため、その状況を可視化したいと思うでしょう。Structured Streamingは、標準の進捗レポートを通じてこれを提供します。

次のマイクロバッチが完了すると、StreamingQueryProgressイベントに再パーティショニング処理の所要時間が含まれます。イベントのdurationMsメトリクス内にある、再パーティショニングの所要時間をミリ秒単位で報告するcontrolBatch.REPARTITIONフィールドを確認してください。

ステートのフットプリントが大きいほど再パーティショニングに時間がかかりますが、ほとんどのワークロードでは数秒しかかからないと予想されます。そのため、大規模なジョブでは、所要時間を把握するためにこのメトリクスを取得する価値があります。これらのイベントの読み取り方法の詳細については、DatabricksでのStructured Streamingクエリの監視を参照してください。

例：クエリのスケールダウン

これを、IDごとのイベントのタンブリングウィンドウカウントというシンプルな集計を例に、具体的に説明しましょう。まずデフォルトの200パーティションで開始し、このワークロードには多すぎると判断して100にスケールダウンします。

まず、デフォルトのパーティション数で現在実行されているクエリです。

しばらくこのストリームを観察した結果、200パーティションは過剰であると判断しました。不要な並列処理のためにコーディネーションのオーバーヘッドが発生しています。クエリを停止し、新しいパーティション数を設定し、同じオプションと同じチェックポイントを使用して再起動します。

再起動されたクエリが立ち上がると、保留中のものがあれば最後に計画されたマイクロバッチを完了し、再パーティショニングを実行してステートを200パーティションから100パーティションに再分配します。その後、すべてのウィンドウとすべての実行中の累計を完全に維持したままカウントを継続します。逆の手順も同様に機能します。より重い負荷の下でスケールアップするには、単に大きな数値を設定します。

同じアプローチがSpark Declarative Pipelines（SDP）にも適用されます。詳細な手順については、ドキュメントのSDPの例を参照してください。

ステート再パーティショニングを使用すべき状況

オンデマンドのステート再パーティショニングは、日常的な操作というよりも、チューニングやスケーリングのためのツールです。以下のような、いくつかの重要な状況で真価を発揮します。 

• 起動後の適切なサイズ調整：最初はストリームが小さく、微調整するほどの価値がなかったため、デフォルトの200パーティションでパイプラインを開始したとします。6か月後、そのパーティション数は失うわけにはいかないチェックポイントに固定されてしまい、しかもそれだけでは不十分になっています。例えば、単一のパイロット地域で開始された不正検知ストリームが、現在はすべての市場をカバーするようになり、200パーティションでは各パーティションが保持するステートが多すぎる状態になっています。オンデマンドの再パーティショニングを使用すれば、既存のチェックポイントを失うことなく、現在の負荷に合わせてパーティション数を増やすことができます。
• ワークロードの変化：ピーク時のトラフィックに合わせてストリームのサイズを決定したとします。例えば、広告入札パイプラインは日中に活発に動き、夜間は静かになります。そのため、日中のピークに合わせて調整された値では、午前3時にほとんどのパーティションがアイドル状態になってしまいます。オンデマンドの再パーティショニングを使用すれば、混雑する時間帯に向けてスケールアップし、それが過ぎたらスケールダウンできるため、最悪のケースではなく実際の負荷に合わせてパーティショニングを行うことができます。
• 履歴データのバックフィル： バックフィルと定常状態の処理では必要なパーティション数が異なりますが、以前はチェックポイントの有効期間中、どちらか一方を選択しなければなりませんでした。例えば、2年分の履歴データを再処理するには、作業を分散させて迅速に完了するために多くのパーティション数が必要ですが、定常状態のトラフィックに戻った後は、同じパーティション数では無駄が多くなります。オンデマンドの再パーティショニングを使用すれば、バックフィル用にスケールアップし、追いついた後は定常状態のサイズにスケールダウンすることができます。これらはすべて、チェックポイントとステートを失うことなく行えます。
• パフォーマンスチューニング：パーティション数は並列処理、ステートサイズ、シャッフルのオーバーヘッドに影響するため、最適な値を予測するのは困難です。例えば、200では少なすぎて、400にすればマイクロバッチのレイテンシが短縮されると考えるかもしれません。しかし、以前のテストではステートの再構築とデータの再処理が必要であり、リソースが無駄になっていました。オンデマンドの再パーティショニングを使用すれば、稼働中のチェックポイントに対してパーティション数を調整し、controlBatch.REPARTITIONやマイクロバッチの所要時間を監視して、推測ではなく測定値に基づいて決定を下すことができます。

変更を行うたびに、1回限りの再パーティショニングによる一時停止を伴う停止と再起動が必要になるため、これは計画的なメンテナンス作業として扱ってください。短時間処理が一時停止しても問題のない時間帯にサイズ変更を計画し、controlBatch.REPARTITIONを監視して所要時間を確認し、クエリが通常の処理リズムに戻るまで様子を見てください。

まとめ

長年、ステートフルなストリーミングクエリのパーティション数は、最初に一度決定した後は二度と変更しないか、あるいはゼロからステートを再構築するために多大なコストを支払うかのどちらかでした。オンデマンドのステート再パーティショニングは、これらの制約を取り除きます。新しいパーティション数全体にステートを安全に再分配することで、開始時のみの決定事項だったものを、ワークロードの必要に応じていつでも見直せる決定事項へと変えることができます。

その結果、長期稼働するストリームの運用者がまさに求めていたものが実現します。つまり、ステートを失うことなく、停止、設定変更、再起動を行うだけで、スケーリングのニーズに基づいてクエリのサイズを適切に調整できる自由が得られます。

オンデマンドのステート再パーティショニングは、RocksDBステートストアプロバイダーを使用するDatabricks Runtime 18以降で利用可能です。詳細なリファレンスについては、ステートフルストリーミングクエリのオンデマンドステート再パーティショニングを参照してください。

(このブログ記事はAI翻訳ツールを使用して翻訳されています) 原文記事