UiPathがDatabricks上でスケーラブルなリアルタイムETLパイプラインを構築した方法

リアルタイムのエージェント型自動化を実現するには、高速で信頼性があり、スケーラブルなデータ基盤が必要です。UiPathでは、MaestroやInsightsのような製品を支えるために、現代的なストリーミングアーキテクチャが必要でした。これにより、エージェント型自動化メトリクスの展開をほぼリアルタイムで可視化することが可能になりました。この取り組みにより、私たちはAzure Databricksでバッチとストリーミングを統一し、Apache Spark™ Structured Streamingを使用して、企業全体でのエージェント型意思決定を支援するコスト効率の良い低遅延分析を可能にしました。

このブログでは、これらの強化の技術的アプローチ、トレードオフ、および影響について詳しく説明します。

UiPath MaestroとUiPath Insightsにとってストリーミングがなぜ重要なのか

UiPathでは、MaestroやInsightsのような製品が、タイムリーで信頼性の高いデータに大きく依存しています。Maestroは、私たちのエージェント型自動化プラットフォームのオーケストレーション層として機能し、リアルタイムのイベントに基づいてAIエージェント、ロボット、人間を調整します。それがシステムトリガーへの反応であったり、長時間実行するワークフローであったり、ヒューマン・イン・ザ・ループのステップを含む場合であったり、Maestroは高速で正確な信号処理に依存して正しい決定を下します。

これらの自動化を監視し、分析する力を持つUiPath Insightsは、別のニーズを追加します：ニアリアルタイムで主要なメトリクスと行動信号をキャプチャし、トレンドを把握し、ROIを計算し、問題の検出をサポートします。

このような結果を達成するためには、反応的なオーケストレーションとリアルタイムの観測性を提供し、低レイテンシーだけでなく、スケーラブルで信頼性があり、維持可能なデータパイプラインアーキテクチャが必要です。そのニーズが私たちにAzure Databricks上のストリーミングアーキテクチャを再考するきっかけとなりました。

ストリーミングデータ基盤の構築

強力な分析とリアルタイムのモニタリングを実現するためには、スケーラブルで信頼性の高いデータパイプラインの基盤が必要です。過去数年間で、新しい製品機能をサポートし、進化するビジネス要件に対応するために、複数のパイプラインを開発し、拡大しました。今、私たちはこれらのパイプラインを最適化し、コストを節約するだけでなく、より良いスケーラビリティを持ち、Maestroのような新しいサービスからのデータに対して、at-least-once配信保証を実現できるかどうかを評価する機会を得ています。

以前のセットアップ（上記参照）はお客様にとってはうまく機能していましたが、改善の余地もありました：

1. バッチパイプラインは最大30分の遅延を引き起こし、複雑なインフラに依存していました。
2. リアルタイムパイプラインはより速いデータを提供しましたが、コストが高かったです。
3. 私たちの最大のデータセットであるRobotlogsでは、履歴処理とリアルタイム処理の両方に対して別々の取り込みと保存パスを維持していました。これにより、重複と非効率が生じました。
4. UiPath Maestroの新しいETLパイプラインをサポートするためには、新しいUiPath製品であるMaestroに対して、at-least-once配信保証を達成する必要がありました。

これらの課題に対処するため、私たちは大規模なアーキテクチャの改革を行いました。Robotlogsのバッチとリアルタイムの取り込みプロセスを単一のパイプラインに統合し、リアルタイムの取り込みパイプラインをよりコスト効率的でスケーラブルに再設計しました。

なぜDatabricks上のSpark Structured Streamingなのか？

パイプラインアーキテクチャを簡素化し、モダナイズするために、私たちは高スループットのバッチワークロードと低レイテンシのリアルタイムデータの両方を処理できるフレームワークが必要でした。Azure Databricks上のSpark Structured Streaming（SSS）は自然な選択でした。

Spark SQLとSpark Coreの上に構築されたStructured Streamingは、リアルタイムデータを無制限のテーブルとして扱い、馴染みのあるSparkバッチ構造を再利用しながら、フォールトトレラントでスケーラブルなストリーミングエンジンの利点を享受します。この統一されたプログラミングモデルは複雑さを減らし、開発を加速しました。

私たちはすでにSpark Structured Streamingを利用して、Databricksで状態を持つストリーム処理を利用するReal-time Alert機能を開発していました。今では、私たちはその機能を拡張して、次世代のReal-time ingestionパイプラインを構築し、低レイテンシ、スケーラビリティ、コスト効率、at-least-once配信保証を達成することができます。

次世代のリアルタイム取り込み

新しいアーキテクチャ（下図参照）は、以前は別々のコンポーネントだったものをDatabricks上のSpark Structured Streamingを使用した統一されたスケーラブルなパイプラインに統合することで、データ取り込みプロセスを劇的に簡素化します：

この新しい設計の中心には、イベントソースから直接読み取る一連のストリーミングジョブがあります。これらのジョブは、パース、フィルタリング、フラット化を行い、最も重要なこととして、各イベントを参照データと結合してデータウェアハウスに書き込む前にエンリッチメントを行います。

これらのジョブはDatabricks Lakeflow Jobsを使用してオーケストレートし、一時的な障害の場合のリトライとジョブ回復を管理します。この効率的なセットアップは、開発者の生産性とシステムの信頼性の両方を向上させます。

この新しいアーキテクチャの利点には以下のようなものがあります：

• コスト効率：インフラストラクチャの複雑さと計算使用量を減らすことでCOGSを節約
• 低遅延： 取り込み遅延は平均約1分で、これをさらに短縮する柔軟性があります
• 将来に備えたスケーラビリティ: スループットはコア数に比例し、無限にスケールアウトすることができます
• データロストなし： Sparkは、at-least-once配信をサポートする障害復旧処理を行います。
  • 今後の開発でのダウンストリームシンクの重複排除により、exactly-once配信を達成することができます
• Spark DataFrame API による迅速な開発サイクル
• シンプルで統一されたアーキテクチャ

低レイテンシ

現在、私たちのストリーミングジョブはマイクロバッチモードで動作し、トリガー間隔は1分です。これは、イベントが私たちのイベントバスに公開される瞬間から、通常、データウェアハウスに27秒で到達し、レコードの95%が51秒以内に、99%が72秒以内に到着することを意味します。

Structured Streamingは設定可能なトリガー設定を提供し、これによりレイテンシを数秒にまで短縮することが可能です。現時点では、コストとパフォーマンスのバランスを考慮して1分間のトリガーを選択しましたが、要件が変更された場合には将来的にこれを下げる柔軟性があります。

スケーラビリティ

Sparkは大量のデータ作業をパーティションに分け、Worker/ExecutorのCPUコアをフルに活用します。各Structured Streamingジョブはステージに分割され、さらにタスクに分割され、各タスクは単一のコアで実行されます。このレベルの並列化により、私たちはSparkクラスタを完全に活用し、増加するデータ量に効率的にスケールすることができます。

メモリ内処理、Catalystクエリ計画、全ステージコード生成、ベクトル化実行などの最適化のおかげで、スケーラビリティ検証では秒間約40,000イベントを処理しています。トラフィックが増加した場合、ソースEvent Busのパーティション数を増やし、さらにワーカーノードを追加することで簡単にスケールアウトできます。これにより、最小限のエンジニアリング努力で将来に備えたスケーラビリティを確保します。

配信保証

Spark Structured Streamingは、そのチェックポイントシステムのおかげで、デフォルトで正確に一度だけ配信を提供します。マイクロバッチごとに、Sparkは各ソースパーティションの進行状況（または「エポック」）とジョブのアプリケーション状態を先行ログとして保存し、状態ストアに保存します。障害が発生した場合、ジョブは最後のチェックポイントから再開し、データが失われたりスキップされたりしないことを保証します。

これは元のSpark Structured Streamingの研究論文で述べられており、一度だけの配信を達成するには以下が必要であると述べています：

1. 入力ソースは再生可能であること
2. 出力シンクは冪等性のある書き込みをサポートします

しかし、しばしば言及されない第三の要件も暗黙のうちに存在します：システムは故障を検出し、適切に対処することができなければなりません。

ここでSparkがうまく機能します - その堅牢な障害回復メカニズムはタスクの失敗、エグゼキュータのクラッシュ、ドライバの問題を検出し、再試行や再起動などの自動的な修正措置を取ることができます。

現在、出力シンクがまだ冪等でないため、少なくとも一度の配信で運用していることに注意してください。将来的に厳密に一度だけの配信の要件がある場合でも、冪等性にさらなるエンジニアリング努力を注げば、それを達成することができるはずです。

生のデータがより良い

また、他の改善もいくつか行いました。すべてのテーブルに共通のrawMessageフィールドを含め、永続化しました。この列は、元のイベントペイロードを生の文字列として保存します。寿司の原則を借りると（ここでは少し違う意味ですが）：生のデータの方が良い。

生データはトラブルシューティングを大幅に簡素化します。何か問題が発生した場合 - 例えば、欠落しているフィールドや予期しない値 - 私たちはすぐに元のメッセージを参照し、ログや上流システムを調査することなく問題を特定できます。この生のペイロードがなければ、データ問題の診断ははるかに困難で遅くなります。

デメリットは、ストレージのわずかな増加です。しかし、安価なクラウドストレージと私たちのウェアハウスのカラム形式のおかげで、これは最小のコストで、クエリのパフォーマンスには影響を与えません。

シンプルでパワフルなAPI

新しい実装は、開発時間を短縮しています。これは主に、分散データ処理に対する高レベルの宣言的抽象を提供するSparkのDataFrame APIのおかげです。過去には、RDDを使用するということは、実行計画について手動で推論し、DAGを理解し、ジョインやフィルターのような操作の順序を最適化することを意味していました。DataFramesは、計算方法ではなく、何を計算したいのかのロジックに焦点を当てることができます。これにより、開発プロセスが大幅に簡素化されます。

これにより、運用も改善されました。失敗したジョブを手動で再実行したり、複数のパイプラインコンポーネント間でエラーを追跡したりする必要はもうありません。簡素化されたアーキテクチャと可動部分が少ないことにより、開発とデバッグが大幅に容易になりました。

UiPath全体でリアルタイム分析を推進

この新しいアーキテクチャの成功は見逃されていません。これはすぐに、UiPath全体のリアルタイムイベント取り込みの新しい基準となりました。UiPath MaestroとInsightsの初期実装を超えて、このパターンは、最先端のイニシアチブを手がけているチームを含む、新しいチームやプロジェクトによって広く採用されています。この広範な採用は、アーキテクチャのスケーラビリティ、効率性、拡張性を証明しており、新しいチームがオンボーディングしやすく、強力なリアルタイム分析機能を持つ新世代の製品を可能にしています。

もし、運用負荷なしにリアルタイムの分析ワークロードをスケールアップしたいのであれば、ここで概説したアーキテクチャは、DatabricksとSpark Structured Streamingによって動力を得て、次世代のAIとエージェントシステムをサポートする準備ができている、実証済みのパスを提供します。

UiPathについて
UiPath (NYSE: PATH)は、エージェント型自動化のグローバルリーダーであり、企業がAIエージェントの全潜在能力を活用して複雑なビジネスプロセスを自動的に実行し、最適化することを可能にしています。UiPath Platform™は、制御されたエージェント機能、開発者の柔軟性、シームレスな統合を独自に組み合わせ、組織がエージェント型自動化を安全かつ自信を持ってスケールアップするのを支援します。セキュリティ、ガバナンス、相互運用性にコミットメントを持つUiPathは、自動化がAIの全潜在能力を発揮して産業を変革する未来への移行を企業が支援します。