Databricks は、ANSI 標準の結合構文をサポートしています。 このページでは、バッチ処理とストリーム処理での結合の違いについて説明します。
注
Databricks では、集合演算子 UNION、INTERSECT、EXCEPT の標準構文もサポートされています。
集合演算子を参照してください。
ストリーミング結合とバッチ結合の違い
Azure Databricks での結合はステートフルまたはステートレスのいずれかです。
すべてのバッチ結合はステートレス結合です。 結果はすぐに処理され、クエリ実行時にデータが反映されます。 クエリが実行されるたびに、指定されたソース データに基づいて新しい結果が計算されます。 「バッチ結合」を参照してください。
2 つのストリーミング データ ソース間の結合はステートフルです。 ステートフル結合では、Azure Databricks はデータ ソースと結果に関する情報を追跡し、結果を繰り返し更新します。 ステートフル結合はオンライン データ処理のための強力なソリューションを提供できますが、効果的に実装するのが難しい場合があります。 これらは、出力モード、トリガー間隔、ウォーターマークに応じて複雑な操作セマンティクスを持ちます。 「ストリーム間の結合」を参照してください。
ストリーム静的結合はステートレスですが、増分データ ソース (ファクト テーブルなど) と静的データ ソース (ゆっくりと変化するディメンション テーブルなど) を結合するための適切なオプションを提供します。 クエリが実行されるたびに両側からのすべてのレコードを結合するのではなく、ストリーミング ソースから新しく受信したレコードのみが現在のバージョンの静的テーブルと結合されます。 「ストリーム静的結合」を参照してください。
バッチ結合
Azure Databricks は、内部結合、外部結合、半結合、反結合、クロス結合などの標準の SQL 結合構文をサポートしています。 JOINを参照してください。
注
Databricks では、具体化されたビューを使って内部結合の結果の増分計算を最適化することをお勧めします。 具体化されたビューを参照してください。
ストリーム同士の結合
2 つのストリーミング データ ソースを結合すると、状態情報の管理と、結果の計算と出力に関する推論に大きな課題が生じる可能性があります。 Databricks では、ストリーム間の結合を実装する前に、ウォーターマークが状態管理にどのような影響を与えるかなど、ステートフル ストリーミングの運用セマンティクスを深く理解することをお勧めします。 次のページを参照してください。
Databricks では、すべてのストリーム間結合の両側にウォーターマークを指定することをお勧めします。 次の結合の種類がサポートされています。
- 内部結合
- 左外部結合
- 右外部結合
- 完全外部結合
- 左半結合
ストリーム ストリーム結合については、Apache Spark Structured Streaming のドキュメントを参照してください。
ストリーム静的結合
注
ストリーム静的結合について説明されている動作は、静的データが Delta Lake を使って格納されていることを前提としています。
ストリーム静的結合では、ステートレス結合を使用して、Delta テーブル (静的データ) の最新の有効なバージョンがデータ ストリームに結合されます。
Azure Databricks がストリーム静的結合でデータのマイクロ バッチを処理する場合、静的 Delta テーブルの最新の有効なバージョンのデータが、現在のマイクロ バッチに存在するレコードと結合されます。 結合はステートレスであるため、透かしを構成する必要がなく、低待機時間で結果を処理できます。 この結合で使用される静的 Delta テーブル内のデータは、緩やかに変化します。
注
実行の間に静的テーブルを更新すると、同じストリーミング データを再処理すると、異なる結果が生成される可能性があります。 ストリーム静的結合の出力は、各マイクロバッチ結合が処理時に最新バージョンの静的テーブルに対して結合されるため、結合の静的側が変更される場合は決定論的ではありません。
以下の例は、ストリーミング bookings テーブルを静的 users テーブルに接続しています。
availableNowトリガーを使用しており、サーバーレスに対応しています。 「構造化ストリーミングのトリガー間隔を構成する」を参照してください。
この例では、出力テーブルとストリームのチェックポイントを main.example_outputに書き、存在しない場合はスキーマとボリュームを作成します。 別の場所で書く場合は、例の上部にある catalog と schema 値を変更してください。
catalog = "main"
schema = "example_output"
spark.sql(f"CREATE SCHEMA IF NOT EXISTS {catalog}.{schema}")
spark.sql(f"CREATE VOLUME IF NOT EXISTS {catalog}.{schema}.checkpoints")
checkpoint_path = f"/Volumes/{catalog}/{schema}/checkpoints/bookings_with_user_info"
streamingDF = spark.readStream.table("samples.wanderbricks.bookings")
staticDF = spark.read.table("samples.wanderbricks.users").select("user_id", "name", "country")
query = (streamingDF
.join(staticDF, "user_id", "inner")
.writeStream
.option("checkpointLocation", checkpoint_path)
.trigger(availableNow=True)
.table(f"{catalog}.{schema}.bookings_with_user_info")
)
Azure Databricks の結合ヒント
Apache Spark では、範囲結合と傾斜結合の結合ヒントの指定がサポートされています。 Azure Databricks ではこれらの結合が自動的に最適化されるため、スキュー結合のヒントは必要ありません。 ヒントを参照してください。
範囲結合のヒントは、結合のパフォーマンスが悪く、不均衡の結合を実行している場合に役立ちます。 たとえば、タイムスタンプ範囲またはクラスタリング ID の範囲での結合が挙げられます。 Azure Databricks での範囲結合の最適化と結合のパフォーマンスの最適化に関するページを参照してください。