エージェント向け：Spark宣言型パイプラインを採用すべき理由

AIエージェントが本番データパイプラインの作成において占める割合が増加しており（Databricks上のSDPパイプラインの70%以上が現在エージェントによって作成されています）、これにより次の2つの課題の重要性が増しています。(1) エージェントの試行錯誤にはコンピュートと時間がかかるため、イテレーションを安価かつ効率的に実行する必要があること、(2) コードレビューが重要な関門となるため、コードが簡潔で、可読性が高く、正確である必要があることです。Apache Spark®には、パイプラインを構築するためのプロシージャル（手続き型）と宣言型の2つのプログラミングモデルが用意されています。これらは、記述内容、実行前にSparkが検証できる内容、およびミスが発生した際コストの面で異なります。同じジョブとエージェントを使用して両者をベンチマークしたところ、宣言型の方がより安全で、低コスト、かつスリムであることがわかりました。

プロシージャルパイプラインは、各データセットをステップバイステップで構築する方法をSparkに指示するコードです。正しく動作するかどうかを確認する唯一の方法は、実行することです。

宣言型パイプラインは、必要なデータセットを記述し、その構築方法の決定をSparkに任せます。これは、LLMにプロンプトを与えて（適切なスキルやツールを使用して）構築方法を判断させるのと似ています。単純なタイポによってどのように失敗するかを示すコードの例を以下に示します。

bronze = spark.read.parquet(RAW)
bronze.write.saveAsTable("bronze")
silver = 
bronze.withColumnRenamed("revenue", 
"revenues")  # typo
silver.write.saveAsTable("silver")
gold = 
spark.read.table("silver").groupBy("revenue")
.sum("revenue")
gold.write.saveAsTable("gold")  
# runtime error

@dp.materialized_view(name="gold")
def gold():
    return 
spark.read.table("silver").groupBy("revenue")
.sum("revenue")

1. 宣言型パイプラインは実行前に検証が可能

バグには2つの種類があります。構造的なバグは、未解決のカラム参照、まだ存在しないテーブル、許可されていない設定変更など、パイプラインを読むことで発見できます。一方、サイレントバグは、実行が完了して誤った結果が出力されて初めて判明します。

宣言型パイプラインでは、Rustのような型付き言語におけるコンパイル時のデータフローの検証と同様に、実行前に構造的なバグをチェックできます。宣言型パイプラインは実行ではなく記述であるため、Sparkはコードを読み取ることで次の3種類の問いに答えることができます。

• すべてのカラム参照が、それが読み込まれるデータセットのスキーマに対して解決されるため、存在しないカラムは実行時エラーではなく拒否されます。
• データセット間のすべての依存関係がグラフ内のプロデューサーと照合されるため、何も宣言していないテーブルへの参照はパイプラインの開始前に検知されます。
• グラフの完全性と順序がチェックされるため、モデルが書き込んだ順序をそのまま信頼するのではなく、エンジンが何を実行するかを決定します。

2. 宣言型パイプラインによるボイラープレートタスクのSparkへのオフロード

プロシージャルのPySparkでは、AIコーディングエージェントが無効な行のフィルタリングやテーブルの結合などのデータ操作を記述します。さらに、パイプラインのステップを正しい順序で実行し、その結果を保存するためのコードも記述します。このサポートコードは、パイプライン間で多くのボイラープレートを繰り返します。

宣言型パイプラインでは、AIエージェントが各データセットとそれを生成するクエリを定義します。実行順序の制御と結果の保存はSparkが処理します。Sparkがボイラープレートを処理するため、エージェントが記述するコードの量は減り、チームのレビューおよび保守の負担も軽減されます。

テスト内容

22のデータエンジニアリングタスクをテストしました。各タスクは実際のチケットのようになっており、問題の説明と、パイプラインが生成する必要があるテーブル、カラム名、データ型が指定されていました。出力がそれらの要件を満たしていれば、エージェントは任意の実装を選択できました。

各タスクについて、本番データに見られるような複雑な値を含む12種類の異なる入力データセットを生成しました。エージェントは、データセットごとにプロシージャルのPySparkを使用したパイプラインと、Spark Declarative Pipelinesを使用したパイプラインをそれぞれ作成しました。これにより、アプローチごとに264回、合計528回のビルドが作成されました（22タスク × 12データセット × 2アプローチ）。すべてのビルドでOpus 4.8を使用しました。

各パイプラインを実行し、その出力を必須テーブルと照合して確認しました。自動採点システムは、どちらのアプローチで結果が生成されたかを知りませんでした。宣言型パイプラインは、最初にSparkのドライランにかけられました。ビルドが失敗した場合、エージェントは上限回数まで修正して再試行することができました。

公開リポジトリから全体を再実行できます。セットアップ、プロンプト、データセット、採点システム、およびすべての数値の背景にある結果が含まれています。

結果

両方の方法でほぼすべてのタスクが完了し、プロシージャルでは264件中255件、宣言型では264件中258件が完了しました。つまり、どちらの方法でも稼働するパイプラインを作成できます。違いは、それ以外のすべての要素、つまり、ミスが表面化したタイミング、修正に必要なコードの量、および出力の一貫性です。以下に宣言型パイプラインの3つのメリットと、その後の2つの注意点について説明します。

1. 実行前の検証により、宣言型パイプラインの時間を約34分の1に短縮

パイプラインのミスは、実行前、実行中、あるいはまったく検知されないかのいずれかで表面化します。プロシージャルパイプラインには実行前ステージがないため、実行前に検知されたミスはありませんでした。構造的なエラーを見つけるために、Sparkは壊れたステップに到達するまでクラスター上でパイプラインを実行する必要がありました。

測定されたビルド全体で、プロシージャルの実行には596エグゼキューター秒がかかったのに対し、宣言型の実行では17秒でした。宣言型コードのドライラン検証により、構造的エラーを早期に検知し、コンピュートコストを34倍削減できました。

2. 宣言型パイプラインのコードサイズがプロシージャルパイプラインの半分に

完了したビルドのうち、宣言型パイプラインのサイズは約半分であり、中央値はプロシージャルパイプラインの131行に対し、64行でした。

また、プロシージャルの実装にはより大きなばらつきが見られました。同じジョブであっても、Sparkのトランスフォーメーションを使用するものもあれば、データをメモリに読み込んで1行ずつ処理するものもありました。どちらもテストデータは通過しましたが、メモリ内処理バージョンはデータが増加すると失敗する可能性があります。ベンチマーク全体で、19件のプロシージャルビルドがこのショートカットを使用しました。宣言型ビルドでは1件も使用されませんでした。

より短く一貫性のあるコードにより、チームのレビュー負担が減り、維持すべきパターンを1つに絞ることができます。

3. ロジックエラーに関しては両者同等

これらが2つのメリットです。テストの実行時に、2つの注意点も確認されました。

まず、ドライランで検知できるのは構造的なミスのみです。サイレントバグ、つまり正常に実行されて誤った答えを返すような最も危険なタイプのミスは検知できません。この点に関して、両者は同等でした。サイレントバグの発生率は互いに約15%の範囲内であり、実際の差とするには小さすぎるギャップでした。宣言型により構造的なミスの検知は容易になりますが、ロジックに関しては効果がないため、引き続きテストとレビューが必要です。

4. 宣言型は事前により多くのコストをかけて後でのコストを削減

宣言型では、ある部分でより多くのコストがかかります。エージェントはパイプラインが健全になるまでドライランの反復と再試行を行うため、クラスター上で後から問題を発見するのではなく、事前により多くの思考をトークン単位で行います。トークンは、パイプラインの作成時に支払われる固定の一度限りのコストであり、データのサイズやパイプラインの実行頻度によって金額は変動しません。無駄なクラスターコンピュートはその逆で、不正な実行のたびに再発し、データ量とともに増加するため、処理するデータが多いほど、不具合のあるビルドのコストがかさみます。増加し続けるコストを回避するために固定のコストを事前に支払うアプローチは、パイプラインとデータが大きくなるほど有利になります。

測定項目の比較

測定された差異の全体像は以下の通りです。

まとめ

本番スケールにおいては、宣言的アプローチが優れています。LLMトークンのコストは約2.3倍になりますが、その引き換えに、失敗するはずのパイプラインにクラスター時間を無駄に費やすことがなくなります。本番スケールにおいて、宣言的アプローチは、LLMトークンのコストがわずかに高くなる代わりに、失敗するパイプラインでのクラスター時間の無駄を確実に防ぐことで有利であることが証明されています。データ量が拡大するにつれて、コンピューティングの無駄は比例して増加し、トークンコストは一定のまま維持されるため、これはますます費用対効果が高くなります。

パイプライン作業のエージェントを評価する場合は、どのエージェントを選ぶかと同じくらい、エージェントがどのようにコードを書くかを重視してください。より優れたエージェントはミスが少なく、宣言的パイプラインはミスを安価にし、コストが発生する前にそれを発見します。

リポジトリには、設計、データ、および完全な来歴が文書化されています。Spark Declarative Pipelinesでエージェントを使い始めるには、公式プログラミングガイドを参照し、PySparkでSDP skillを使用してください。

FAQ

これを機能させるためにAIエージェントを信用する必要がありますか？

いいえ、信用する必要がないことがポイントです。宣言的エージェントが記述するのは宣言的パイプラインコード（必要なテーブルの説明）のみであり、実行前にSpark自体がその説明をチェックします。チェッカーはプラットフォームの一部です。信頼するのはモデルではなく、パイプラインに対するSparkの評価です。エージェントが自分でコードを実行したり、データを読み取ったりすることは一切ありません。

これはSpark Declarative Pipelinesだけのものですか、それともより広範に適用されますか？

これは宣言的アプローチにより広範に適用されます。「ここに必要なものがあります」と「それを構築してください」を分離できる場所ならどこでも、最初の部分を単体でチェックできます。SQL自体、インフラストラクチャ・アズ・コードツール、スキーマファーストAPIでも同様の構成が見られます。Spark Declarative Pipelinesはベンチマークを行ったものですが、その理屈は、そのように構築されたあらゆるツールに当てはまります。

宣言的エージェントは2.3倍のトークンを使用します。それは問題外ではありませんか？

現実的なパイプラインサイズであれば問題ありません。Opus 4.8のリスト料金に基づくと、2.3倍という数字は、中央値の宣言的ビルドあたりトークンで約0.50ドルになり、手続き型の場合は0.26ドルになります。手続き型の実行が1回失敗すると約521エグゼキューター秒が無駄になりますが、拒否された宣言的実行の場合は約0.5秒で済むため、数回の失敗した実行を回避するだけで多くの追加トークンをカバーでき、データ量が増加するにつれてその差はさらに広がります。トークンの価格は利用するモデルサービスによって変動しますが、コンピューティングコストはより安定しており、計画を立てやすくなります。

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