Databricksにおける大規模なdbtプロジェクトのスケーリングと運用：パフォーマンス、可視性、デバッグに関するIFCOデータチームの事例

概要

IFCOは、数億個ものクレートやパレットを擁する、世界最大規模の再利用可能なパッケージングプールを運営しています。世界全体で2,000人以上の従業員を擁するIFCOは、ドイツ国内で350人以上を雇用しており、その大半はミュンヘン近郊のプーラッハにあるグローバル本社で働いています。同社のビジネスは循環型のプーリングサービスです。リユーザブル・プラスチック・コンテナ（RPC）が、生産者や包装業者から配送センターや小売業者へと生鮮食品を運び、その後、IFCOのサービスセンターに戻されて洗浄、分類され、再び50カ国以上に送り出されます。

すべてのクレートとパレットはライフサイクルを通じて追跡され、ビジネス運営の指標となるKPI（サイクルタイム、紛失、破損、洗浄コスト、プール規模など）に反映されます。何十億もの未加工の追跡イベントを信頼できるKPIに変換することは、次の3つの理由から困難です。まずデータ量が膨大であること、次に一部のデータが予測困難な形で遅れて到着すること、そして遅延データが到着した際に、パイプラインがすでに報告済みの過去の履歴を修正せざるを得なくなることです。

本記事では、IFCOのデータプラットフォームチームがDatabricksのForward Deployed Engineeringと協力し、このパイプラインをより高速かつ低コストに抑えた方法を紹介します。変換ロジックはdbt内に維持されます。これはDatabricks上で実行され、各インクリメンタル（増分）設定が、具体的なDelta Lakeの書き込み動作（どの列でデータをクラスタリングするか、書き込みがターゲットテーブルのどの程度に影響を及ぼすか、行をマージするか置換するかなど）にマッピングされます。適切なデータ粒度でこれらの設定を正しく行うことにより、コアとなるセマンティックレイヤージョブの日次の実行時間が60%以上短縮され、IFCOはコストのかかる夜間のフルリフレッシュを廃止することができました。

IFCOとデータ課題の様相

クレートは年に何度も回収、充填、出荷、返却、洗浄、再利用されるため、IFCOはすべての資産がどこにあり、何が起きたかを把握する必要があります。IFCOはセマンティックレイヤーを導入し、さまざまな追跡シグナル（洗浄ラインを通過する際のバーコードスキャン、ドックドアでのRFID読み取り、GPS位置、近くのBluetoothビーコン、温度を報告するバッテリー駆動のトラッカーなど）を、資産アクティビティのガバナンスが効いた1つのビューに統合しました。（本記事全体を通して、「セマンティックレイヤー」とは、未加工の追跡イベントをビジネスKPIに変換する、これらのガバナンスが効いたdbtモデルを指します。）これを困難にしているのは、次の3つの特性です。

• スケール（規模）： 1日に数十億件の追跡イベントが流れ込みます。集約処理によって、各資産からの繰り返しのpingがはるかに少ないアクティビティ行に圧縮されますが、KPIが読み取るテーブルは依然として非常に大きいため、これらをゼロから再構築するには多大なコストがかかります。
• 予測不可能な遅延到着： ほとんどの観測データは想定された時間枠内に収まりますが、一部のフィードは不定期なスケジュールで数週間、場合によっては数ヶ月遅れることがあります。今日手元にあるデータが昨日の出来事の全容であると仮定するパイプラインは、誤った履歴を静かに報告してしまいます。
• 履歴の整合性確保： 資産のアクティビティは一連のシーケンスであるため、遅延した観測データは単に隙間を埋めるだけではありません。資産のタイムラインの途中にスキャンデータを挿入すると、それ以降のすべてについてパイプラインがすでに導き出した結論（資産が次にどこへ移動したか、サイクルがいつ開始したか、どのKPIバケットに分類されたかなど）が変わってしまいます。したがって、遅延到着が発生すると、単に新しい行を追加するだけでなく、パイプラインはすでに公開済みの状態を再計算せざるを得なくなります。目標は、毎晩すべてを再処理することなく、迅速に適切な予測値を出力し、遅延データが到着するにつれて真の値へと収束させることです。

大規模なインクリメンタル処理

dbtのインクリメンタルモデルの内部は、一連 of Deltaの読み取りおよび書き込み動作であり、今回の最大の成果は「各実行で処理する行数を極力減らし、可能な限り早い段階で削減する」という1つの原則から得られました。最初にして最大のレバーは読み取りそのものです。実行時に実際に必要なファイルと変更された資産のみをスキャンします。読み取りを回避したすべての行は、ダウンストリームにおけるコストのかかる資産ごとのソート、シャッフル、書き込み処理に到達することがないからです。以下に示す各手法は、特定のDeltaの動作に変換される通常のdbt構成（config）です。

フィルターや結合（Join）の基準となる列でクラスタリングする。 Liquid clusteringを、各モデルのクエリ実行時の粒度（資産アクティビティの場合は、資産とイベント日付）に設定することで、エンジンはファイルをスキャンする代わりにスキップできます。これが、次に説明する2つの手法を機能させる鍵となります。

インクリメンタル戦略を慎重に選択する。 この戦略によって各実行時の書き込み方法が決まります。選択の基準となるのは、「各行に固定のキーがあるか」、そして「行をその場で更新するか、それともグループごと一括で置換するか」という2つの問いです。キーが存在し、重複排除が頻発するアップサート（upsert）では、merge（マージ）がデフォルトとなります。実際的な粒度（資産アクティビティの場合は、asset_idとevent_date_time）をキーとすることで、一括の「削除して再挿入（delete-and-reinsert）」では不可能な次の2つの処理が実現します。

等価結合（equi-join）の述語によって、動的ファイルプルーニングが有効になります。これにより、入力バッチ内のキー値に基づいて、一致するデータが含まれる可能性のないターゲットファイルがスキップされるため、書き込みは変更対象のスライスのみに限定されます。（(DBT_INTERNAL_DEST and DBT_INTERNAL_SOURCEは、dbtが生成するステートメントにおける、ターゲットテーブルと入力バッチのエイリアスです。）マージが一致するのと同じキーでクラスタリングを行うことで、このプルーニングを厳密に維持できます。また、各行のサロゲートハッシュを比較するmatched_conditionである行ハッシュガードにより、実際には変更されていない行の再書き込みをスキップします。これにより、書き込みが節約され、ダウンストリームの変更フィードがクリーンに保たれます。

delete+insertは代替策です。これは、キーによって行のグループ全体を削除し、再挿入します。実行時にグループを1つの単位として再派生させ、行に一致させるための固定の識別子がない場合は、こちらの方がシンプルです。ただし、何も変更されていない場合でもグループ全体を再書き込みするというコストが発生します。データ量が非常に膨大な場合は、想定だけで判断せず、これら2つのベンチマークを測定する価値があります。

書き込みを直近の時間枠に制限する。 同じ述語メカズムには、もう1つの用途があります。ファイルプルーニングのための等価結合の代わりに、時間制限を設けて書き込みを直近のデータに制限します。これにより、データ量の多いアップストリームモデルにおいて、MERGEはテーブル全体ではなく、デスティネーションの直近のスライスに対してマッチングを行います。

述語はイベントの発生時刻ではなく、行が取り込まれた（インジェストされた）時刻をキーとするため、数ヶ月前のイベントであっても、最近取り込まれたものであれば捕捉されます。時間枠は、データが到着してからこのジョブが処理するまでのギャップをカバーできる広さにするだけで十分です。この枠を狭く設定しすぎると、遅延データはエラーを出すことなく静かにスキップされ、処理されなくなります。

変更された部分のみを再計算する。 モデルは、インジェストのウォーターマークから特定された、新規データまたは遅延データの影響を受けた資産に対象を絞って処理を行います。また、書き込み時よりも広い時間枠で読み取りを行うため、フルリフレッシュを行うことなく遅延イベントを捕捉できます。

Deltaをクリーンに保つ。 負荷の高いインクリメンタルテーブルでは、最適化された書き込み（optimized writes）と自動コンパクション（auto-compaction）を有効にするか、テーブルのメンテナンスをPredictive Optimizationに任せることで、頻繁なマージによって小さなファイルが大量に発生し、読み取りパフォーマンスが低下するのを防ぎます。

重要なのは、これらを適切な粒度で適用した上で、実際のクエリプランから、エンジンがサイレントスキャンを行わずに実際にプルーニングを実行しているかを確認することです。

具体的な事例：観測データの資産アクティビティへの集約

セマンティックレイヤーで最も負荷の高いモデルは、あらゆる追跡技術からの観測データを、資産ごとの位置情報を考慮した単一のストリームに集約するモデルです。このモデルは、資産ごとにパーティショニングされ、イベント時刻順にソートされたウィンドウ関数を使用して、資産が実際にいつ移動したかを算出します。観測データに明示的な位置情報が含まれていない場合は、Databricks SQLに組み込まれたH3関数にフォールバックします。この関数は、各緯度・経度を六角形のグリッドセルにマッピングするため、「同じ場所」の判定が、地理的距離の計算を繰り返すのではなく、セルIDとグリッド距離の低コストな比較で済むようになります。この処理は、実行時間を最も多く消費する最大の要因となっていました。

最初のステップは、最適化することではなく、実際に何が実行されているかを確認することでした。この違いは極めて重要です。dbt compileは、参照（refs）が解決されたモデルのSELECTをレンダリングしますが、インクリメンタルモデルの場合、それはDatabricksが実行する実際のステートメントではありません。コンパイルされたSELECTの裏で、dbtは一時ビュー、デスティネーションテーブルのスキャン、およびテーブルへの最終的な書き戻しといった、より大きな処理を生成して実行します。時間とメモリがどこで消費されているかを突き止める唯一の方法は、コンパイルされたSQLではなく、クエリ履歴から実際に実行されたクエリプランをステージごとに読み解くことです。

そのようにして確認したプランは、致命的な状態を示していました。モデルは何十億もの行をスキャンし、数百ギガバイトのデータをディスクにスピル（退避）させ、実行時間の約85%を資産ごとの単一のウィンドウソートとシャッフルに費やしていました。実質的に、実行するたびにテーブル全体を再構築していたのです。これには3つの原因がありました。

• 「変更された」アセットのセットが縮小することはありませんでした。アップストリームのタイムスタンプがソースから引き継がれるのではなく、実行のたびに再生成されていたため、ほぼすべてのアセットが新規のように見えていました。すべてが変更ありと見なされると、インクリメンタル実行はいつの間にかフルリフレッシュになってしまいます。
• アセットごとの再計算に制限がありませんでした。ウィンドウ関数が各アセットの履歴をすべて遡って参照していたため、実際に変更されたセットがごくわずかであっても、ソート処理に何年分もの履歴が引きずり込まれていました。
• ウィンドウステージが、最終クエリでまったく使用されない列を計算していました。これには、出力全体が破棄される2つ目の将来予測ウィンドウも含まれていました。

それぞれの修正は、その原因に直接対応しています。実際の取り込みタイムスタンプをアップストリームモデル経由で引き継ぎ、変更されたセットが真に新しいデータを反映するようにする。再計算を直近のウィンドウに制限する。未使用の列と将来予測ウィンドウを削除する。モデルがクエリされる粒度でクラスター化する。そして最後に、グラフ全体をサーバーレスコンピュート上でモデルごとの並列タスクとして実行する（次のセクションを参照）。これらにより、コアジョブの実行時間は60%以上（ほぼ3分の2）短縮され、KPIを正しく維持するために必要だった毎晩のフルリフレッシュが不要になりました。

1回限りのデバッグセッションから、再現可能なスキルへ

上記の診断（コンパイルされたSQLではなく、実際に実行されたプランを読み取る、読み取り側と書き込み側の両方を確認する、各症状を根本原因まで追跡する）は、統合モデルに特有のものではありません。これは、Databricks上の低速なインクリメンタルモデルに対して、あらゆるエンジニアが実行する一連の手順です。この手順をパッケージ化したものが「スキル」です。これはAIエージェントが必要に応じて実行するプレイブックであり、手順を覚えているエンジニアの数ではなく、モデルの数に合わせて診断をスケールさせることができます。

このスキルは、先ほどの実例をステップバイステップで再現します。インクリメンタルモデルではこれらが異なるステートメントになるため、dbt compileの出力ではなく、クエリ履歴から実際のステートメントファミリーを取得します。増幅は書き込み側にしか現れないため、実行の双方向、つまりスキャン側のメトリクス（プルーニングされたファイル、読み取られた行、スピル）と書き込み側のメトリクス（書き込まれた行と削除された行）の両方を読み取ります。次に、統合モデルで見つかったものと同じ3つの障害クラスをチェックします。縮小しない変更セット（引き継がれずに再生成されたアップストリームのタイムスタンプ）、制限のないアセットごとの再計算（ルックバック制限のないウィンドウ）、および無駄な処理（計算されたもののダウンストリームで読み取られなかった列やウィンドウパス）です。各チェックは直感ではなく、メトリクスやプランのシグナルに基づいています。

出力は、勝手に修正するのではなくレポートとして提示されます。各検出事項は、その証拠（スキャンされた行、スピルバイト、プランノード）とともに、提案される変更案とペアで提示され、承認されるまでモデルには何も適用されません。承認されると、修正の根拠となったものと同じ事前/事後のメトリクスが次の実行時に再測定されます。そのため、このスキルは修正が機能したと決めつけるのではなく、フィードバックループを完結させます。

得られるメリットは、目新しさではなく一貫性です。統合モデルの実行時間の原因となった3つの要因（再生成されたタイムスタンプ、制限のないウィンドウ、未使用の列）は、ありふれたものであり、負荷がかかっている状況では見落としがちです。これらを検出するためのスキルの実行は、繰り返し行うコストがほぼかからず、実行時間が60%増加して誰かがエスカレーションしなければならなくなる前に、次のモデルで同種の課題を発見できます。

Databricks Jobsでの大規模なdbtプロジェクトの実行

Databricks Workflows（Lakeflow Jobs）は、dbtをファーストクラスのタスクタイプとして扱います。これにより、dbtプロジェクトを、管理された1つのワークフロー内で、取り込みやダウンストリームのステップと並べてスケジュール、実行、監視でき、再試行やアラートを共有できます。最もシンプルな方法では、プロジェクト全体を単一のdbtタスクとして実行します。これは機能しますが、ブラックボックスになってしまいます。1つのモデルが失敗するとジョブ全体が失敗し、個々のモデルを確認、再実行、または監視する方法がありません。この規模では、これは運用上のリスクとなります。

解決策は、dbtグラフを、モデル、テスト、シード、スナップショットごとに1つずつ、個別のDatabricksタスクとして実行することです。IFCOは、スタンドアロンのオープンソースライブラリであるdatabricks-dbt-factory（MITライセンス、GitHubおよびPyPIで公開）を使用して、そのグラフを生成します。これはdbtマニフェストとジョブテンプレートを読み取り、ノードごとにタスクを持つDatabricks Asset Bundleジョブを生成します。タスクごとの粒度が効果を発揮するのは、各タスクの起動コストが低い場合のみであり、これは次の3つのメカニズムに集約されます。

• ノートブックタスクタイプ。小さな共有ランナーノートブックが、Python APIを通じてdbtをトリガーします。dbtがSQLを生成してSQLウェアハウスに送信するため、ノートブックのコンピュートはデータを処理しません。
• ベース環境。ベース環境とは、事前構築されたサーバーレスPython環境のスナップショットです。これにはすでにdbt-databricksが含まれているため、各タスクはここから開始され、新規タスクで発生するはずのpip installをスキップできます。
• マニフェストインジェクション。事前構築されたマニフェストがdbtに直接渡されるため、パース処理がスキップされ、各タスクはプライベートなローカルディレクトリにアーティファクトを書き込みます。多くのタスクが同時に実行される大規模なプロジェクトでは、これによりプロジェクトの再読み込みに数分かかるか、すぐに有益な処理を開始できるかの違いが生まれます。

オーバーヘッドを小さく抑えることで、ファンアウトは運用に必要なものを提供します。タスクレベルの可視性、失敗したモデルとその依存関係のみを対象とした再実行、モデルごとのロギング、アラート、テスト、そして拡張可能なランナー（シークレットのロード、実行へのgit SHAのタグ付け、数行でのSlackへの投稿など）です。すべてが、パスを認識するGitHub Actionsマトリックスを介してDatabricks Asset Bundlesとしてデプロイされ、モデルごとの実行時間とコストは、クエリタグとシステムテーブルからアラート付きのダッシュボードに追跡されます。そのため、パフォーマンスの低下は毎月の請求書ではなく、1日以内に明らかになります。

大規模なテストと品質管理

効率性が向上しても、数値がいつの間にか壊れてしまっては意味がありません。そのため、品質は期待するものではなく、強制されるものです。すべてのモデルには所有者と一意性テストが設定されています。プライマリーキーは、エラーレベルの重要度で一意性と非NULL（not null）がテストされます。実際のロジック（ウィンドウ関数、複数の結合、複雑なマクロ）を持つモデルには、単体テストが必要です。dbt-bouncerは、Databricksダイアレクト上のsqlfluffとともに、これらに違反するコミットをブロックし、外部の消費者が読み取るレイヤーにはコントラクトが適用されます。

テスト自体のコストにも同様の規律が適用されます。ビューベースのチェックは実行のたびにビューを再計算するため、ビューに対するチェックはマテリアライズされるか、より少ないパスにまとめられます。基本的なチェックは列制約になり、テストはインクリメンタルデータにスコープされます。ローカルでは、開発者は本番環境のマニフェストを参照するため、変更されたモデルのみがビルドされ、アップストリームは本番環境から読み取ります。CIでは、マージ前に変更されたモデルに対して単体テストとサンプリングデータテストが実行されます。

ベンチマーク

今後の展望

現在のパイプラインはバッチ処理です。データの取り込みは1日に1回行われ、その上でセマンティックレイヤーが実行され、早期の見積もりを出力し、遅延データが到着するにつれて収束させます。今回の取り組みの中で推奨された3つの作業により、これをさらに進化させ、ほぼリアルタイムのKPIを実現する道が開かれます。

• Change Data Feedを使用して、変更されたデータのみを消費する。Delta LakeのChange Data Feedを使用すると、ダウンストリームモデルは入力を再スキャンする代わりに、アップストリームで変更された行のみを読み取ることができます。これを統合モデルとその依存関係に適用することで、照合処理が「直近のウィンドウをスキャンする」から「移動した正確な行を処理する」へと縮小されます。これは、再計算を制限した後の自然な次のステップです。
• 低遅延の取り込み。1日1回のロードは、ダウンストリームのアーキテクチャを変更することなく、ストリーミング取り込みパス、Auto Loader、または低遅延のLakeflow Connectコネクタに移行できます。これだけで、データの鮮度が1日単位から分単位へと向上します。
• 宣言型ストリーミング変換。同じ変換ロジックを、毎晩のバッチとしてではなく、ストリームを読み取るLakeflow Declarative Pipelineとして継続的に実行し、イベントの到着に合わせてステートフル処理でアセットごとの照合を行うことができます。

決定的な要因は、技術ではなくビジネス上のニーズです。メトリクスを翌朝ではなく、数分以内に最新の状態にする必要がある場合、このパスは同じ管理されたテーブル上で、同じdbt定義のロジックを使用してそれを提供します。1日1回のリフレッシュで十分な場合は、バッチパイプラインの方がすでに低コストな解決策となっています。

まとめ

このソリューションの形は、明確な役割分担にあります。変換ロジックはモジュール化されテストされた状態でdbt内に維持される一方、データは単一の Unity Catalog ガバナンスモデルのもとでオープンな Delta Lake フォーマットに保持されます。これにより、テーブルを再構築するたびにリネージやアクセス制御が失われることはなく、特定のクエリエンジンに依存することもありません。このdbtロジックは、スケール向けに構築された Databricks の機能（リキッドクラスタリング、Delta インクリメンタル書き込み、動的ファイルプルーニング、H3地理空間関数など）へとコンパイルされます。コンパイルされたSQLではなく実際のクエリプランから診断を行い、コストの高いソートやシャッフルに到達する行数を削減します。さらに、プロジェクトをモデルごとのタスクグラフとして実行することで、運用面での可視性を高め、低いオーバーヘッドで安全に再実行できるようにします。最大の成果は、クラスターを大きくしたことではなく、作業量を減らしたこと（アクセスする行数を減らし、再計算するアセットを減らし、テーブルの再構築頻度を大幅に下げること）によってもたらされました。

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