Delta Live TablesとApache Kafkaによる低レイテンシストリーミングデータパイプライン

公開日: 2022年8月9日

製品2 min read

Delta Live Tables (DLT) は、信頼性の高いデータパイプラインの作成にシンプルな宣言型アプローチを採用し、バッチ処理とストリーミングデータの両方で基盤となるインフラストラクチャを大規模に完全に管理する初の ETL フレームワークです。多くのユースケースでは、ほぼリアルタイムのデータから導き出される実行可能なインサイトが必要とされます。Delta Live Tables は、Apache Kafka、AWS Kinesis、Confluent Cloud、Amazon MSK、または Azure Event Hubs のようなイベントバスから直接データを取り込むことで、低レイテンシーを実現し、そのようなユースケースをサポートする低レイテンシーのストリーミングデータパイプラインを可能にします。

この記事では、DLT と Apache Kafka を使用する方法を、ストリームを取り込むために必要な Python コードとともに説明します。推奨されるシステムアーキテクチャについて説明し、関連する DLT 設定についても検討します。

ストリーミングプラットフォーム

イベントバスまたはメッセージバスは、メッセージプロデューサーとコンシューマーを分離します。一般的なストリーミングのユースケースは、Web サイトをナビゲートするユーザーからのクリックストリームデータの収集であり、すべてのユーザーインタラクションは Apache Kafka のイベントとして保存されます。Kafka からのイベントストリームは、リアルタイムのストリーミングデータ分析に使用されます。複数のメッセージコンシューマーは、Kafka から同じデータを読み取り、そのデータを使用してオーディエンスの関心、コンバージョン率、および直帰理由について学習できます。ユーザーインタラクションからのリアルタイムのストリーミングイベントデータは、請求データベースに保存されている実際の購入と相関させる必要がある場合もよくあります。

Apache Kafka

Apache Kafka は、一般的なオープンソースのイベントバスです。Kafka は、イベントの追記専用分散ログであるトピックの概念を使用しており、メッセージは一定期間バッファリングされます。Kafka のメッセージは、消費された後も削除されませんが、無期限に保存されるわけでもありません。Kafka のメッセージ保持期間はトピックごとに設定可能で、デフォルトは 7 日間です。期限切れのメッセージは最終的に削除されます。

この記事は Apache Kafka を中心としていますが、ここで説明する概念は、他の多くのイベントバスまたはメッセージングシステムにも適用されます。

ストリーミングデータパイプライン

データフローパイプラインでは、Delta Live Tables とその依存関係は、標準的な SQL の Create Table As Select (CTAS) ステートメントと DLT キーワード「live」で宣言できます。

DLT を Python で開発する場合、@dlt.table デコレータを使用して Delta Live Table を作成します。パイプラインのデータ品質を確保するために、DLT は Expectations を使用します。これは、無効なレコードに対するパイプラインの動作を定義するシンプルな SQL 制約句です。

ストリーミングワークロードは予測不可能なデータ量で発生することが多いため、Databricks はデータフローパイプラインに拡張オートスケーリングを採用し、不要なインフラストラクチャをシャットダウンすることでコストを削減しながら、エンドツーエンドのレイテンシーを最小限に抑えます。

Delta Live Tables は、パイプラインの実行ごとに、正しい順序で、正確に 1 回、完全に再計算されます。

対照的に、ストリーミング Delta Live Tables はステートフルであり、増分計算され、最後のパイプライン実行以降に追加されたデータのみを処理します。ストリーミングライブテーブルを定義するクエリが変更された場合、新しいデータは新しいクエリに基づいて処理されますが、既存のデータは再計算されません。ストリーミングライブテーブルは常にストリーミングソースを使用し、Kafka、Kinesis、または Auto Loader のような追記専用ストリームでのみ機能します。ストリーミング DLT は Spark Structured Streaming の上に構築されています。

非常に大量のデータ量と低レイテンシー要件を持つワークロードなど、複数のストリーミングパイプラインを連鎖させることができます。

ストリーミングエンジンからの直接取り込み

Python で記述された Delta Live Tables は、Spark Structured Streaming を使用して Kafka のようなイベントバスから直接データを取り込むことができます。コンプライアンスの問題を回避し、コストを削減するために Kafka トピックの保持期間を短く設定し、Delta が提供する安価で弾力的で管理可能なストレージのメリットを享受できます。

パイプラインの最初のステップとして、複雑な変換によって重要なデータが失われるのを避けるために、データをそのままブロンズ（生）テーブルに取り込むことをお勧めします。他の Delta Table と同様に、ブロンズテーブルは履歴を保持し、GDPR およびその他のコンプライアンスタスクを実行できるようにします。

DLT パイプラインを Python で記述する場合、@dlt.table アノテーションを使用して DLT テーブルを作成します。Python でストリーミング DLT をマークするための特別な属性はありません。ストリームにアクセスするには、spark.readStream() を使用するだけです。Kafka トピックからデータを消費する kafka_bronze という名前の DLT テーブルを作成する例コードは次のとおりです。

pipelines.reset.allowed

イベントバスは通常、一定期間経過するとメッセージを期限切れにしますが、Delta は無期限の保持を目的として設計されていることに注意してください。

これにより、DLT パイプラインのフルリフレッシュを実行したときに、Kafka 上のソースデータがすでに削除されているという状況が発生する可能性があります。この場合、メッセージングプラットフォームからすべての履歴データをバックフィルできず、DLT テーブルにデータが不足する可能性があります。データの損失を防ぐには、次の DLT テーブルプロパティを使用します。

pipelines.reset.allowed=false

pipelines.reset.allowed を false に設定すると、テーブルへのリフレッシュは防止されますが、テーブルへの増分書き込みや、テーブルへの新しいデータのフローは防止されません。

チェックポインティング

経験豊富な Spark Structured Streaming 開発者であれば、上記のコードにチェックポインティングがないことに気づくでしょう。Spark Structured Streaming では、正常に処理されたデータの進捗情報を永続化するためにチェックポインティングが必要であり、障害発生時には、このメタデータを使用して失敗したクエリを中断した箇所から正確に再開します。

チェックポイントは Spark Structured Streaming での正確な一度の保証による障害回復に必要ですが、DLT は手動設定や明示的なチェックポインティングなしに状態を自動的に処理します。

DLT パイプラインでの SQL と Python の混在

DLT パイプラインは複数のノートブックで構成できますが、1 つの DLT ノートブックは、SQL または Python のいずれかで完全に記述する必要があります（他の Databricks ノートブックとは異なり、1 つのノートブックで異なる言語のセルを持つことができます）。

SQL を好む場合は、Python で 1 つのノートブックで Apache Kafka からのデータ取り込みをコーディングし、その後、データパイプラインの変換ロジックを別のノートブックで SQL で実装できます。

スキーママッピング

メッセージングプラットフォームからデータを読み取る際、データストリームは不透明であり、スキーマを提供する必要があります。

以下の Python の例は、フィットネストラッカーからのイベントのスキーマ定義と、Kafka メッセージの値部分がそのスキーマにどのようにマッピングされるかを示しています。

メリット

DLT でメッセージブローカーから直接ストリーミングデータを読み取ることで、アーキテクチャの複雑さが最小限に抑えられ、メッセージブローカーから直接ストリーミングされ、中間ステップがないため、エンドツーエンドのレイテンシーが短縮されます。

クラウドオブジェクトストアを介したストリーミング取り込み

特定のユースケースでは、Apache Kafkaからデータをオフロードし、ストリーミングデータをクラウドオブジェクトストアに保存したい場合があります。Databricksワークスペースでは、クラウドベンダー固有のオブジェクトストアをDatabricks Files System (DBFS) を介して、クラウドに依存しないフォルダーとしてマッピングできます。データがオフロードされると、Databricks Auto Loader がファイルをインジェストできます。

Auto Loaderは、単一行のSQLコードでデータをインジェストできます。JSONファイルをDLTテーブルにインジェストする構文を以下に示します（読みやすくするために2行に折り返されています）。

Auto Loader自体はストリーミングデータソースであり、新しく到着したすべてのファイルは正確に1回処理されることに注意してください。そのため、rawテーブルのストリーミングキーワードは、データがそのテーブルにインクリメンタルに取り込まれることを示しています。

ストリーミングデータをクラウドオブジェクトストアにオフロードすると、システムアーキテクチャに余分なステップが追加されるため、エンドツーエンドのレイテンシーが増加し、追加のストレージコストが発生します。イベントデータをクラウドオブジェクトストアに書き込むKafkaコネクタを管理する必要があり、運用上の複雑さが増すことに注意してください。

したがって、Databricksは、ベストプラクティスとして、上記のように、Spark Structured Streamingを使用してDLTからイベントバスデータに直接アクセスすることを推奨しています。

その他のイベントバスまたはメッセージングシステム

この記事はApache Kafkaを中心に展開していますが、ここで説明する概念は、他のイベントバスまたはメッセージングシステムにも適用できます。DLTは、Databricks Runtimeが直接サポートするあらゆるデータソースをサポートしています。

Amazon Kinesis

Kinesisでは、メッセージを完全に管理されたサーバーレスストリームに書き込みます。Kafkaと同様に、Kinesisはメッセージを永続的に保存しません。Kinesisのデフォルトのメッセージ保持期間は1日です。

Amazon Kinesisを使用する場合は、上記のストリーミング取り込みのPythonコードでformat("kafka")をformat("kinesis")に置き換え、option()でAmazon Kinesis固有の設定を追加します。詳細については、Spark Structured StreamingドキュメントのKinesis統合セクションを確認してください。

Azure Event Hubs

Azure Event Hubsの設定については、Microsoftの公式ドキュメントと、記事Delta Live Tablesレシピ：Azure Event Hubsからの消費を確認してください。

概要

DLTは、ETLの「T」以上のものです。DLTを使用すると、ストリーミングソースとバッチソースから簡単にデータをインジェストし、Databricks Lakehouse Platform上でデータをクレンジングおよび変換して、あらゆるクラウドでデータ品質を保証できます。

Apache Kafkaからのデータは、DLTノートブックからPythonで直接Kafkaブローカーに接続することでインジェストできます。Kafkaストリーミングレイヤーのソースデータが期限切れになった場合でも、完全なパイプラインリフレッシュのためにデータ損失を防ぐことができます。

はじめに

Databricksの顧客の場合は、はじめにガイドに従ってください。リリースノートを読んで、このGAリリースに含まれるものについて詳しく学んでください。まだDatabricksの顧客でない場合は、無料トライアルにサインアップしてください。詳細なDLT価格については、こちらをご覧ください。

データとAIサミット2022の発表やアップデートについて話し合っているデータ愛好家の仲間たちがいるDatabricksコミュニティで会話に参加しましょう。学び、ネットワークを作りましょう。

最後に、サミットからのDive Deeper into Data Engineeringセッションをお楽しみください。このセッションでは、Twitterライブストリーム、Auto Loader、SQLでのDelta Live Tables、Hugging Faceセンチメント分析を使用した別のストリーミングデータ例のコードを説明します。

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