Lakehouse、Spark Declarative Pipelines、PySparkの選び方と、これらを組み合わせるべきタイミング
によって Rafael Aielo による投稿
チームには何百ものストアドプロシージャ、いくつかのスケジューラがあり、権限はロールやスキーマに分散しており、クラウドデータウェアハウスの更新期限が迫っています。何を最初に移行すべきか、誰も意見が一致しません。すべてをPySparkで書き直したい人もいれば、SQLをそのまま移行して完了にしたい人もいます。議論の中で見落とされているのは、コードとともに移行されるメタデータ、リネージ、権限、そして移行の過程でこれらを統合する機会です。
どちらの極端なアプローチもうまくいきません。データウェアハウスの移行に成功するチームは、各ワークロードを個別に評価し、そのタスクに最適なツールを選択しています。この記事では、選択のための意思決定フレームワークを提案します。いつLakehouse（Databricks SQL）、Spark Declarative Pipelines、またはPySparkを使用すべきか、そして計画で行き詰まることなく成果を出せるように作業を段階的に進める方法について説明します。
Databricksでは、ETLパイプラインを主に3つの方法で移行でき、これらは多くの場合組み合わせて使用されます。
これは、SQLを多用するチームにとって最も直接的なパスです。シンプルなものから複雑なものまで幅広く対応します。これはSQLウェアハウス上で実行され、デフォルトでPhotonによって高速化されており、ANSIおよびSpark SQL（%sql）と完全に互換性があります。変動しやすい、または予測不可能なワークロードにはServerlessを選択してください（高速起動、ゼロへのスケール、秒単位の課金）。安定したワークロードや、特定のネットワーク制御やコスト制御が必要な場合はClassicを選択してください。
シンプルなSQLタスク：
ロジックに制御フロー（条件分岐）、ループ、変数、エラー処理、またはパラメータ駆動型の実行が必要な場合、ストアドプロシージャがその手続き型レイヤーを提供します。これらはUnity Catalogを介して管理され、パラメータを指定してWorkflowsから呼び出すことができます。
大まかな目安として、レガシーコードが単一のSQLステートメントである場合は、SQLタスクとして移行します。手続き型ロジック（変数、ループ、パラメータ、エラー処理）が含まれている場合は、Unity Catalogで管理され、Workflowsから呼び出し可能なストアドプロシージャにラップします。元のシステムで必要だったという理由だけで、シンプルなSQLをプロシージャにラップしないでください。
Lakeflowの一部であるこの機能は、異なるアプローチをとります。パイプラインが何を生成すべきかを宣言すると、エンジンが実行順序、再試行、スケーリングを処理します。組み込みのデータ品質制約、自動的な依存関係の解決、そして同一の定義内でのバ ッチとストリーミングの統合を利用できます。
内部では、Enzymeが、派生テーブルを段階的に更新するか、完全に再計算するかを決定します。自動スケーリングにより、手動でのチューニングなしで、データ量の変化に合わせて容量が調整されます。Blockなどの企業は、利用規模の拡大に伴うパイプラインのオーケストレーションを簡素化するために、この宣言型モデルを活用しています。
これらは完全な制御を提供します。ジョブクラスター上で実行され、SQLウェアハウスや宣言型パイプラインに適さないワークロードを処理します。
ワークロードに複雑なビジネスロジック、MLの特徴量エンジニアリング、API統合、またはカスタム検証が必要な場合は、PySparkを使用してください。以下の例では、Unity Catalogに登録されたモデルを使用してトランザクションをスコアリングしています。
言語はSQLのままであるものの、ワークロードがSQLウェアハウスの適正を超える可能性がある場合（非常に大きなテーブル、重いシャッフル、パーティショニング、ブロードキャストジョイン、またはキャッシュを明示的に制御したい長期実行のバッチETLなど）は、ノートブック内のSpark SQLを使用してください。
大規模なジョイン、集計、ウィンドウ関数、巨大な列指向テーブルのスキャンなど、計算負荷の高いSQLまたはDataFrameの処理を行う場合は、ジョブクラスターでPhotonを有効にしてください。Photonは、Pandas UDF（Arrowベース）を含むこれらのパターンをコード変更なしで高速化する、ネイティブなベクトル化エンジンです。行単位のPython UDFが大部分を占める場合、データセットが小さい場合、またはジョブが純粋なI/Oである場合は、Photonをスキップしてください。
ノートブックは、SDPでの取り込み、ノートブックタスクでのエンリッチメントといったハイブリッドパイプラインにも適しています。
以下の表はチームでの議論の出発点であり、厳格なルールではありません。
| 基準 | Lakehouse（タスクとストアドプロシージャ） | Spark Declarative Pipelines | PySpark + Spark SQLノートブック |
|---|---|---|---|
| チームのプロフィール | SQL中心、DBA、DWエンジニア | マネージドパイプラインを構築するデータエンジニアおよびSQLチーム | Python/Sparkデベロッパー、MLエンジニア |
| ロジックのタイプ | SQL ETL：単一ステートメント用のシンプルなタスク、手続き型ロジック用のストアドプロシージャ | 宣言型パイプライン、CDC、SCD | 複雑なロジック、カスタムUDF、ML準備 |
| SQL移行スピード | 高（SQL ANSIライクなワークロードの場合） | 中：パイプラインの再設計が必要だが、SQLの再利用が可能 | 可変 ：大幅なリファクタリングが必要になる可能性あり |
| パイプラインのオーケストレーション | WorkflowsでのSQLタスクまたはCALLプロシージャ | パイプラインに組み込み | Workflowsでのノートブックタスク |
| バッチ vs. ストリーミング | 主にバッチ | 統合されたバッチとストリーミング | Structured Streamingによるバッチおよびストリーミング |
| データ品質 | 手動のSQLチェック | 宣言型の制約 | コード内でのカスタム検証 |
列からお使いのチームを、行からワークロードの複雑さを選択してください。交差するセルが、開始すべき場所を示しています。
| ワークロードの複雑さ | SQLファーストのチーム | ハイブリッドチーム | コードファーストのチーム |
|---|---|---|---|
| 低 （バッチロード、集計、MERGE） | WorkflowsでのSQLタスク | SQLタスクまたはSDP | PySparkまたはSDP |
| 中 （マルチステップパイプライン、CDC、データ品質） | ストアドプロシージャまたはSDP | SDP | SDPまたはPySpark |
| 高 （ML準備、カスタムUDF、API、高密度のビジネスロジック） | SDP ＋ PySparkアシスト | 取り込み用のPySpark ＋ SDP | PySpark |
「すべてに対してどのアプローチをとるか」を決めるのではなく、各フェーズで「 次に何をすべきか」を決定します。
フェーズ1 — 評価。レガシーデータウェアハウスから、CPU時間、実行時間、頻度、ソーステーブルとターゲットテーブルなどのメトリクスを収集します。ワークロードを複雑さに基づいて分類します。可能な場合は移行ツールを使用して、価値と難易度でスコアリングされたインベントリを作成します。このデータの場所はソースによって異なります。Teradataでは、DBC.QryLogをクエリします。SQL Serverでは、sys.dm_exec_query_statsを使用します。Oracleでは、AWRレポートを使用します。Snowflakeでは、QUERY_HISTORYを使用します。詳細は異なる場合があります。統合ツールが導入されている場合は、そのメタデータを活用してテーブル間の関係を特定したり、LLMを利用してこのリネージを構築したりできます。出力されるのはマップであり、書き換え計画ではありません。目標は同じです。リソース消費量と依存関係のレベルでワークロードをランク付けし、どこから始めるべきかを把握することです。適切に行えば、移行ツールを使用することで、この評価は手動のスク リプト作成に何週間もかけることなく、数日で完了します。
フェーズ2 — クイックウィン。移行リスクが低く、ビジネス上の注目度が高いユースケースを組み合わせたワークロードを選択します。これは、変換が容易な重いSQLジョブから始めることや、新しいプラットフォームを早い段階で関係者に示すことができるレポートパイプラインから始めることを意味する場合があります。シンプルなステートメントはWorkflowsのSQLタスクになります。手続き型ロジックはストアドプロシージャになります。初期の変換には、トランスパイラやAI支援による変換を使用します。両方のシステムを並行して実行し、行数、チェックサム、サンプルレコードを比較します。重要なのは、技術的および組織的な信頼を築くことです。
たとえば、Walgreensは段階的な移行によってオンプレミスのTeradataを廃止し、現在はレイクハウス上で1秒あたり約40,000件のデータイベントを処理しており、約9,000店舗のサプライチェーン最適化を支援しています。
フェーズ3 — 近代化。次に、近代化する価値のあるパイプラインを再設計します。対象となるのは、データ品質の制約とリネージによって手動チェックを削減できるフロー、ストリーミングテーブルやCDCの恩恵を受けるバッチジョブ、マテリアライズドビューによって複雑さが軽減されるパイプライン、そして以前は個別のツールに存在していたメタデータ、権限、監査がUnity Catalogの下に統合されたものです。一般的なパターンは、新しいパイプラインが検証に合格するまで並行して 実行する間、レガシーなプロシージャをフォールバックとして維持することです。近代化されたパイプラインは、多くの場合、バッチ処理時間を数時間から数分に短縮し、個別のDQツールの必要性を排除します。
フェーズ4 — 最適化。古いDWの制限を回避するためだけに存在していた冗長なETLパイプラインを統合します。ロジックがシンプルになる場合は、複雑なホットスポットをPySparkに移行します。統合されたエンジンが手に入った今、バッチとストリーミングの境界を再検討します。ここで移行の成果が現れます。レガシープラットフォームは停止し、冗長なパイプラインは排除され、アーキテクチャは2つではなく1つのシステムで実行されます。
移行ツールは機械的な作業を自動化しますが、アーキテクチャの決定に代わるものではありません。代表的な3つの役割は以下の通りです。
実用的なアプローチ：初期変換の60〜80%をツールに任せ、エンジニアの時間は実際に近代化したいパターンに充てます。これにより、技術的負債をそのまま移行してしまうことを回避できます。
移行を成功させるには、単にコードを変換するだけでなく、スタンドアロンのスケジューラサーバー、カスタムDQフレームワーク、個別のリネージおよびメタデータツール、ベンダー固有のストアドプロシージャコンパイラ、独自に作成した検証ハーネスなどのシステムを積極的に廃止します。これらのシステムが停止し、課金が止まるまで、移行は完了しません。
SQL ETLがエンジン、レイヤー、ツール間で断片化したままであれば、データがオープンフォーマットで保存されていても、プラットフォームは断片化したままになります。
データパイプラインを移行するための唯一の正しい方法というものはありません。レイクハウスなら迅速に移行できます。シンプルなロジックにはシンプルなタスクを、手続き型の制御が必要な場合はストアドプロシージャを使用します。SDPは、品質とリネージが組み込まれた最新のETLパイプラインを提供します。PySparkを使用する場合でもSpark SQLを使用する場合でも、残りの処理はNotebooksが処理します。作業を段階的に進め、クイックウィンから始め、利用可能なすべてのアクセラレータを活用してください。
技術的なウォークスルーについてはDatabricks移行ガイドを参照するか、Databricks Free Editionでご自身でお試しください。
(このブログ記事はAI翻訳ツールを使用して翻訳されています) 原文記事
ブログを購読して、最新の投稿を受信トレイにお届けします。