Zerobus IngestとLakebaseでほぼリアルタイムのアプリケーションを構築

IoT、クリックストリーム、アプリケーションテレメトリーからのイベントデータは、Databricksデータインテリジェンスプラットフォームと組み合わせることで、重要なリアルタイム分析とAIを強化します。従来、このデータを取り込むには、データソースとlakehouseの間でいくつかのデータホップ（メッセージバス、Sparkジョブ）が必要でした。これにより、運用上のオーバーヘッドやデータの重複が増加し、専門知識が必要となるため、レイクハウスがこのデータの唯一の宛先である場合は一般的に非効率です。

このデータがレイクハウスに格納されると、変換およびキュレーションされ、下流の分析ユースケースに利用されます。しかし、チームは多くの場合、この分析データを運用ユースケースに提供する必要があり、これらのカスタムアプリケーションの構築は骨の折れるプロセスになる可能性があります。チームは、専用のOLTPデータベースインスタンス（ネットワーキング、モニタリング、バックアップなど）のような、必要不可欠なインフラストラクチャコンポーネントをプロビジョニングし、維持する必要があります。さらに、リアルタイムアプリケーションで再表示するために、分析データをデータベースに取り込むリバースETLプロセスを管理する必要があります。そのためには、チームはレイクハウスから外部の運用データベースにデータをプッシュするための追加パイプラインを構築する必要があります。これらのパイプラインは、開発者が設定および維持する必要のあるインフラストラクチャに追加され、ビジネス向けのアプリケーションを構築するという主要な目標から注意をそらすことになります。

Databricksは、lakehouseへのデータの取り込みと、オペレーショナルなワークロードをサポートするためのゴールドデータの提供の両方を、どのように簡素化するのでしょうか？

Zerobus Ingest と Lakebase の登場

Zerobus インジェストについて

Lakeflow Connect の一部である Zerobus Ingest は、イベントデータをレイクハウスに直接プッシュする効率的な方法を提供するAPIsセットです。Zerobus Ingestは、シングルシンクメッセージバスレイヤーを完全に排除することで、インフラストラクチャを削減し、運用を簡素化し、大規模でほぼリアルタイムの取り込みを実現します。そのため、Zerobus Ingest を使用すると、これまで以上に簡単にデータの価値を引き出すことができます。

データを生成するアプリケーションは、データを書き込むターゲットテーブルを指定し、メッセージがテーブルのスキーマに正しくマッピングされるようにし、Databricksにデータを送信するストリームを開始する必要があります。Databricks側では、APIがメッセージとテーブルのスキーマを検証し、データをターゲットテーブルに書き込み、データが永続化されたことをクライアントに確認応答として送信します。

Zerobus Ingestの主なメリット:

• 効率化されたアーキテクチャ: 複雑なワークフローやデータの重複が不要になります。
• 大規模なパフォーマンス：ほぼリアルタイムの取り込み（最大5秒）をサポートし、何千ものクライアントが同じテーブルに書き込むことができます（クライアントあたり最大100MB/秒のスループット）。
• データインテリジェンスプラットフォームとの統合：不正検知のためのMLflowなどのアナリティクスツールやAIツールをデータに直接適用できるようにすることで、価値実現までの時間を短縮します。

Lakebaseについて

Lakebaseは、Databricks Platformに組み込まれた、フルマネージド、serverless、スケーラブルなPostgresデータベースであり、分析およびAIのユースケースを支える同じデータ上で直接実行される、低レイテンシーの運用およびトランザクションワークロード向けに設計されています。

コンピュートとストレージを完全に分離することで、迅速なプロビジョニングと弾力性のあるオートスケールが実現します。LakebaseのDatabricks Platformとの統合は、従来のデータベースとの大きな差別化要因です。なぜなら、Lakebaseは複雑なカスタムデータパイプラインを必要とせずに、LakehouseのデータをリアルタイムアプリケーションとAIの両方で直接利用可能にするからです。エンタープライズアプリケーションやエージェントワークロードを強化するために、データベースの作成、クエリーのレイテンシー、同時実行性の要件を満たすように構築されています。最後に、開発者はコードのようにデータベースを簡単にバージョン管理し、ブランチを作成できます。

Lakebaseの主なメリット:

• 自動データ同期：複雑な外部パイプラインを必要とせずに、スナップショット、スケジュール、または継続的にLakehouse（分析レイヤー）からLakebaseにデータを簡単に同期する機能
• Databricksプラットフォームとの統合：Lakebaseは、Unity Catalog、Lakeflow Connect、Spark Declarative Pipelines、Databricks Appsなどと統合されています。
• 統合された権限とガバナンス: オペレーショナルデータと分析データに対する一貫したロールと権限の管理。ネイティブの Postgres 権限は、Postgres プロトコルを介して引き続き維持できます。

これらのツールを組み合わせることで、顧客は複数のシステムから Delta テーブルに直接データをインジェストし、大規模なリバース ETL のユースケースを実装できます。次に、これらのテクノロジーを使用してニアリアルタイムのアプリケーションを実装する方法を見ていきましょう！

ほぼリアルタイムのアプリケーションを構築する方法

実践的な例として、フードデリバリー会社「Data Diners」の管理スタッフが、ドライバーの活動と注文の配送をリアルタイムで監視できるアプリケーションを使えるように支援します。現在、彼らにはこの可視性が欠けており、配送中に問題が発生した場合に対処する能力が制限されています。

なぜリアルタイムアプリケーションは価値があるのか？

• 運用状況の把握：管理者は、各ドライバーがどこにいるか、現在の配送がどのように進んでいるかを即座に確認できます。これにより、注文の遅延やドライバーが支援を必要とする際の死角が少なくなります。
• 問題の軽減: リアルタイムの位置情報とステータスデータにより、配車係は遅延が発生した場合にドライバーのルートを変更したり、優先順位を調整したり、顧客に積極的に連絡したりすることができ、配達の失敗や遅延を削減できます。

データインテリジェンスプラットフォーム上でZerobus Ingest、Lakebase、Databricks Appsを使用してこれを構築する方法を見ていきましょう！

アプリケーションアーキテクチャの概要

このエンドツーエンドのアーキテクチャは、次の 4 つのステージで構成されています。(1) データプロデューサーは Zerobus SDK を使用して、イベントを Databricks Unity Catalog の Delta テーブルに直接書き込みます。(2) 継続的な同期パイプラインが、更新されたレコードを Delta テーブルから Lakebase Postgres インスタンスにプッシュします。(3) FastAPI バックエンドは、WebSocket 経由で Lakebase に接続し、リアルタイムの更新をストリーミングします。(4) Databricks Apps上に構築されたフロントエンド アプリケーションが、エンドユーザー向けにライブデータを可視化します。

データプロデューサーから始めると、ドライバーの電話にあるデータダイナーアプリは、注文を配達する途中のドライバーの位置（緯度と経度の座標）に関するGPSテレメトリデータを出力します。このデータはAPIゲートウェイに送信され、最終的に取り込みアーキテクチャの次のサービスにデータを送信します。

Zerobus SDK を使用すると、API ゲートウェイからターゲットテーブルにイベントを転送するクライアントを迅速に作成できます。ターゲットテーブルがほぼリアルタイムで更新されるため、継続的な同期パイプラインを作成してlakebaseテーブルを更新できます。最後に、Databricks Apps を活用することで、WebSocket を使用して Postgres からリアルタイムの更新をストリームする FastAPI バックエンドと、ライブ データフローを可視化するフロントエンド アプリケーションをデプロイできます。

Zerobus SDKが登場する以前は、ストリーミングアーキテクチャでは、データがターゲットテーブルに格納されるまでに複数のホップを経由する必要がありました。私たちのAPIゲートウェイは、データをKafkaのようなステージングエリアにオフロードする必要があり、トランザクションをターゲットテーブルに書き込むためにはSpark Structured Streamingが必要だったでしょう。唯一の宛先がレイクハウスであることを考えると、これらすべてが不要な複雑さを生み出しています。上記のアーキテクチャは、Databricks Data Intelligence Platformが、データ取り込みからリアルタイムアナリティクス、インタラクティブなアプリケーションの実装まで、エンドツーエンドのエンタープライズアプリケーション開発をいかに簡素化するかを示しています。

次のステップ

前提条件：必要なもの

• Lakebase: AWSとAzureで一般提供を開始。
• Zerobus Ingest：AWSおよびAzureで一般提供中
• Databricks Apps：Databricks Appsを作成するための権限があることを確認してください。

ステップ1：Databricks Unity Catalogでターゲットテーブルを作成する

クライアントアプリケーションによって生成されるイベントデータは、Delta テーブルに格納されます。以下のコードを使用して、目的のカタログとスキーマにそのターゲットテーブルを作成します。

ステップ2: OAUTHを使用して認証する

ステップ3：Zerobusクライアントを作成し、ターゲットテーブルにデータを取り込む

以下のコードは、Zerobus APIを使用してテレメトリイベントデータをDatabricksにプッシュします。

Change Data Feed (CDF) の制限と回避策

現在のところ、Zerobus Ingestはチェンジデータフィードをサポートしていません。チェンジデータフィードを使用すると、DatabricksはDeltaテーブルに書き込まれた新しいデータの変更イベントを記録できます。これらの変更イベントは、挿入、削除、または更新です。これらの変更イベントは、その後 Lakebase で同期されたテーブルを更新するために使用できます。Lakebase にデータを同期してプロジェクトを続行するには、ターゲットテーブルのデータを新しいテーブルに書き込み、そのテーブルでチェンジデータフィード を有効にします。

ステップ 4: Lakebase をプロビジョニングし、データをデータベースインスタンスに同期する

アプリを強化するために、この新しいCDF対応テーブルからLakebaseインスタンスにデータを同期します。ほぼリアルタイムのダッシュボードをサポートするために、このテーブルを継続的に同期します。

UI で以下を選択します：

• 同期モード：低レイテンシー更新のための継続的同期
• 主キー：table_primary_key

これにより、アプリは最小限の遅延で最新のデータを反映します。

注: Databricks SDK を使用して、プログラムで同期パイプラインを作成することもできます。

foreachライターによるリアルタイムモード

DeltaからLakebaseへの継続的な同期には10～15秒のラグがあるため、より低いレイテンシーが必要な場合は、ForeachWriterライター経由でリアルタイムモードを使用し、DataFrameからLakebaseテーブルにデータを直接同期することを検討してください。これにより、データはミリ秒単位で同期されます。

GithubのLakebase ForeachWriterコードを参照してください。

ステップ5：FastAPIまたは任意のフレームワークでアプリを構築する

データが Lakebase に同期されたので、コードをデプロイしてアプリを構築できます。この例では、アプリはLakebaseからイベントデータを取得し、それを使用して、食品配達中のドライバーの活動を追跡するためのほぼリアルタイムアプリケーションを更新します。Databricksでのアプリ構築についてさらに詳しく知るには、Get Started with Databricks Apps docsをお読みください。 

関連リソース

特定のニーズに合わせて独自のアプリケーションを構築するための、その他のチュートリアル、デモ、ソリューションアクセラレータをご覧ください。

• エンドツーエンドアプリケーションの構築: リアルタイムのセーリングシミュレーターは、Python SDKとREST API、Databricks Apps、Databricks Asset Bundlesを使用して、セーリングボートのフリートを追跡します。ブログを読む
• デジタルツイン ソリューションの構築：Databricks Apps と Lakebase を使用して、運用効率を最大化し、リアルタイムの知見を加速させ、予測メンテナンスを行う方法を学びます。ブログを読む

Zerobus Ingest、Lakebase、およびDatabricks Appsの詳細については、技術ドキュメントをご覧ください。Databricks Apps CookbookやCookbook Resource Collectionもご覧いただけます。

まとめ

IoT、クリックストリーム、テレメトリ、および同様のアプリケーションは、毎日何十億ものデータポイントを生成し、それらはいくつかの業界にわたる重要なリアルタイムアプリケーションを強化するために使用されます。そのため、これらのシステムからのインジェストを簡素化することが最も重要です。Zerobus Ingestは、これらのシステムからイベントデータを直接レイクハウスにプッシュする合理的な方法を提供し、同時に高いパフォーマンスを保証します。これはLakebaseとうまく連携し、エンドツーエンドのエンタープライズアプリケーション開発を簡素化します。

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