パイプラインによるデータの変換

この記事では、DLT を使用してデータセットの変換を宣言し、クエリロジックを通じてレコードを処理する方法を指定する方法について説明します。また、DLT パイプラインを構築するための一般的な変換パターンの例も含まれています。

データフレームを返す任意のクエリに対してデータセットを定義できます。Apache Spark の組み込み操作、UDF、カスタムロジック、MLflow モデルを DLT パイプラインの変換として使用できます。DLT パイプラインにデータを取り込んだら、アップストリームソースに対して新しいデータセットを定義して、新しいストリーミングテーブル、マテリアライズドビュー、およびビューを作成できます。

DLT でステートフル処理を効果的に実行する方法については、「ウォーターマークを使用して DLT でステートフル処理を最適化する」を参照してください。

ビュー、マテリアライズドビュー、ストリーミングテーブルを使用する場合

パイプラインクエリを実装するときは、効率的で保守性を確保するために、最適なデータセットタイプを選択します。

ビューを使用して次のことを行うことを検討してください。

必要な大規模または複雑なクエリを、管理しやすいクエリに分割します。
エクスペクテーションを使用して中間結果を検証します。
ストレージとコンピュートのコストを削減し、保持する必要のない結果を得ることができます。テーブルはマテリアライズされるため、追加の計算リソースとストレージリソースが必要になります。

マテリアライズドビューは、次の場合に使用を検討してください。

複数のダウンストリームクエリがテーブルを消費します。ビューはオンデマンドでコンピュートされるため、ビューがクエリされるたびにビューが再コンピュートされます。
他のパイプライン、ジョブ、またはクエリはテーブルを消費します。ビューは具体化されていないため、同じパイプラインでのみ使用できます。
開発中にクエリの結果を表示する必要があります。テーブルは具体化され、パイプラインの外部で表示およびクエリを実行できるため、開発中にテーブルを使用すると、計算の正確性を検証するのに役立ちます。検証後、実体化を必要としないクエリをビューに変換します。

次の場合は、ストリーミングテーブルの使用を検討してください。

クエリーは、継続的または段階的に増加するデータソースに対して定義されます。
クエリ結果はインクリメンタルにコンピュートする必要があります。
パイプラインには、高いスループットと低いレイテンシが必要です。

注記

ストリーミングテーブルは、常にストリーミングソースに対して定義されます。また、 APPLY CHANGES INTO でストリーミングソースを使用して、CDC フィードからの更新を適用することもできます。「 APPLY CHANGES API: DLTによるチェンジデータキャプチャの簡素化」を参照してください。

ターゲットスキーマからテーブルを除外する

外部消費を目的としない中間テーブルを計算する必要がある場合は、 TEMPORARY キーワードを使用して、中間テーブルがスキーマにパブリッシュされないようにすることができます。一時テーブルは、DLT セマンティクスに従ってデータを格納および処理しますが、現在のパイプラインの外部からアクセスしないでください。一時テーブルは、それを作成するパイプラインの有効期間中保持されます。次の構文を使用して、テンポラリ・テーブルを宣言します。

SQL
Python

SQL
CREATE TEMPORARY STREAMING TABLE temp_table
AS SELECT ... ;

Python
@dlt.table(
  temporary=True)
def temp_table():
  return ("...")

ストリーミングテーブルとマテリアライズドビューを 1 つのパイプラインに結合

ストリーミングテーブルは、 Apache Spark 構造化ストリーミングの処理保証を継承し、新しいローが変更されるのではなく、常にソーステーブルに挿入される追加専用データソースからのクエリを処理するように構成されています。

注記

デフォルトでは、ストリーミングテーブルには追加専用データソースが必要ですが、ストリーミングソースが更新または削除が必要な別のストリーミングテーブルである場合は、skipChangeCommits フラグを使用してこの動作をオーバーライドできます

一般的なストリーミングパターンでは、ソースデータを取り込んでパイプラインに初期データセットを作成します。これらの初期データセットは一般にブロンズテーブルと呼ばれ、多くの場合、単純な変換を実行します。

対照的に、パイプライン内の最終テーブル (一般にゴールドテーブルと呼ばれます) では、多くの場合、複雑な集計や APPLY CHANGES INTO 操作のターゲットからの読み取りが必要になります。これらのオペレーションは本質的に追加ではなく更新を作成するため、ストリーミングテーブルへの入力としてはサポートされていません。これらの変換は、マテリアライズドビューに適しています。

ストリーミングテーブルとマテリアライズドビューを 1 つのパイプラインに混在させることで、パイプラインを簡素化し、コストのかかる生データの再取り込みや再処理を回避し、効率的にエンコードおよびフィルター処理されたデータセットに対して複雑な集計を行うための SQL を最大限に活用できます。次の例は、このタイプの混合処理を示しています。

注記

これらの例では、 Auto Loader を使用してクラウドストレージからファイルを読み込みます。 Auto LoaderUnity Catalog対応パイプラインでを含むファイルをロードするには、外部ロケーションを使用する必要があります。DLT での Unity Catalog の使用の詳細については、「 DLT パイプラインでの Unity Catalog の使用」を参照してください。

Python
SQL

Python
@dlt.table
def streaming_bronze():
  return (
    # Since this is a streaming source, this table is incremental.
    spark.readStream.format("cloudFiles")
      .option("cloudFiles.format", "json")
      .load("gs://path/to/raw/data")
  )

@dlt.table
def streaming_silver():
  # Since we read the bronze table as a stream, this silver table is also
  # updated incrementally.
  return spark.readStream.table("streaming_bronze").where(...)

@dlt.table
def live_gold():
  # This table will be recomputed completely by reading the whole silver table
  # when it is updated.
  return spark.readStream.table("streaming_silver").groupBy("user_id").count()

SQL
CREATE OR REFRESH STREAMING TABLE streaming_bronze
AS SELECT * FROM STREAM read_files(
  "gs://path/to/raw/data",
  format => "json"
)

CREATE OR REFRESH STREAMING TABLE streaming_silver
AS SELECT * FROM STREAM(streaming_bronze) WHERE...

CREATE OR REFRESH MATERIALIZED VIEW live_gold
AS SELECT count(*) FROM streaming_silver GROUP BY user_id

Auto LoaderJSONを使用してからGoogle Cloud Storage ファイルを段階的に取り込む方法について詳しくは、こちらをご覧ください。

ストリーム静的結合

Stream-static 結合は、主に静的ディメンションテーブルを使用して追加専用データの連続ストリームを非正規化する場合に適しています。

パイプラインが更新されるたびに、ストリームの新しいレコードが静的テーブルの最新のスナップショットと結合されます。ストリーミングテーブルの対応するデータが処理された後に静的テーブルにレコードが追加または更新された場合、完全な更新が実行されない限り、結果のレコードは再計算されません。

トリガー実行用に構成されたパイプラインでは、静的テーブルは更新が開始された時点の結果を返します。連続実行用に構成されたパイプラインでは、テーブルが更新を処理するたびに、静的テーブルの最新バージョンがクエリされます。

次に、ストリーム静的結合の例を示します。

Python
SQL

Python
@dlt.table
def customer_sales():
  return spark.readStream.table("sales").join(spark.readStream.table("customers"), ["customer_id"], "left")

SQL
CREATE OR REFRESH STREAMING TABLE customer_sales
AS SELECT * FROM STREAM(sales)
  INNER JOIN LEFT customers USING (customer_id)

集計を効率的に計算

ストリーミングテーブルを使用して、count、min、max、sum などの単純な分布集計や、平均や標準偏差などの代数集計を増分的に計算できます。 Databricks では、 GROUP BY country 句を含むクエリなど、グループの数が制限されているクエリの増分集計をお勧めします。更新のたびに、新しい入力データのみが読み取られます。

増分集計を実行する DLT クエリの記述の詳細については、「ウォーターマーク付きのウィンドウ集計の実行」を参照してください。

DLT パイプラインで MLflow モデルを使用する

注記

Unity Catalog 対応パイプラインで MLflow モデルを使用するには、preview チャンネルを使用するようにパイプラインを構成する必要があります。 current チャンネルを使用するには、 Hive metastoreにパブリッシュするようにパイプラインを設定する必要があります。

MLflow でトレーニングされたモデルは、DLT パイプラインで使用できます。MLflow モデルは Databricks では変換として扱われ、Spark データフレーム入力に作用し、結果を Spark データフレームとして返します。 DLT はに対してデータセットを定義するため、データフレーム Apache Sparkを使用するワークロードをわずか数行のコードでMLflow DLT に変換できます。MLflow の詳細については、「生成AI エージェントの MLflow」と「ML モデルのライフサイクル」を参照してください。

MLflow モデルを呼び出す Python ノートブックが既にある場合は、 @dlt.table デコレーターを使用し、変換結果を返すように関数が定義されていることを確認することで、このコードを DLT に適応させることができます。DLTは MLflow by デフォルトをインストールしませんので、%pip install mlflowと一緒にMLFlowライブラリをインストールし、ノートブックの上部にmlflowとdltをインポートしていることを確認してください。 DLT 構文の概要については、「 Python を使用したパイプラインコードの開発」を参照してください。

DLT で MLflow モデルを使用するには、次の手順を実行します。

MLflow モデルの実行 ID とモデル名を取得します。実行 ID とモデル名は、MLflow モデルの URI を構築するために使用されます。
URI を使用して、MLflow モデルを読み込むための Spark UDF を定義します。
MLflow モデルを使用するには、テーブル定義で UDF を呼び出します。

次の例は、このパターンの基本的な構文を示しています。

Python
%pip install mlflow

import dlt
import mlflow

run_id= "<mlflow-run-id>"
model_name = "<the-model-name-in-run>"
model_uri = f"runs:/{run_id}/{model_name}"
loaded_model_udf = mlflow.pyfunc.spark_udf(spark, model_uri=model_uri)

@dlt.table
def model_predictions():
  return spark.read.table(<input-data>)
    .withColumn("prediction", loaded_model_udf(<model-features>))

完全な例として、次のコードでは、ローンリスクデータでトレーニングされた MLflow モデルを読み込む loaded_model_udf という名前の Spark UDF を定義しています。予測を行うために使用されるデータ列は、引数として UDF に渡されます。 loan_risk_predictionsテーブルでは、loan_risk_input_dataの各行の予測が計算されます。

Python
%pip install mlflow

import dlt
import mlflow
from pyspark.sql.functions import struct

run_id = "mlflow_run_id"
model_name = "the_model_name_in_run"
model_uri = f"runs:/{run_id}/{model_name}"
loaded_model_udf = mlflow.pyfunc.spark_udf(spark, model_uri=model_uri)

categoricals = ["term", "home_ownership", "purpose",
  "addr_state","verification_status","application_type"]

numerics = ["loan_amnt", "emp_length", "annual_inc", "dti", "delinq_2yrs",
  "revol_util", "total_acc", "credit_length_in_years"]

features = categoricals + numerics

@dlt.table(
  comment="GBT ML predictions of loan risk",
  table_properties={
    "quality": "gold"
  }
)
def loan_risk_predictions():
  return spark.read.table("loan_risk_input_data")
    .withColumn('predictions', loaded_model_udf(struct(features)))

手動による削除または更新を保持する

DLT を使用すると、テーブルからレコードを手動で削除または更新し、更新操作を実行してダウンストリームテーブルを再計算できます。

デフォルトでは、DLT はパイプラインが更新されるたびに入力データに基づいてテーブルの結果を再計算するため、削除されたレコードがソースデータから再ロードされないようにする必要があります。pipelines.reset.allowed table プロパティを false に設定すると、テーブルへの更新は防止されますが、テーブルへの増分書き込みや新しいデータがテーブルに流入するのを防ぐことはできません。

次の図は、2 つのストリーミングテーブルを使用した例を示しています。

raw_user_table ソースから生のユーザーデータを取り込みます。
bmi_table raw_user_tableからの体重と身長を使用してBMIスコアを段階的にコンピュートします。

raw_user_tableからユーザーレコードを手動で削除または更新し、bmi_tableを再計算する場合。

データ保持図

次のコードは、 pipelines.reset.allowed テーブルプロパティを false に設定して raw_user_table の完全更新を無効にし、意図した変更が時間の経過と共に保持されるようにする方法を示しています。これにより、パイプラインの更新が実行されるとダウンストリームテーブルが再計算されます。

SQL
CREATE OR REFRESH STREAMING TABLE raw_user_table
TBLPROPERTIES(pipelines.reset.allowed = false)
AS SELECT * FROM STREAM read_files("/databricks-datasets/iot-stream/data-user", format => "csv");

CREATE OR REFRESH STREAMING TABLE bmi_table
AS SELECT userid, (weight/2.2) / pow(height*0.0254,2) AS bmi FROM STREAM(raw_user_table);

ビュー、マテリアライズドビュー、ストリーミングテーブルを使用する場合​

ターゲットスキーマからテーブルを除外する​

ストリーミングテーブルとマテリアライズドビューを 1 つのパイプラインに結合​

ストリーム静的結合​

集計を効率的に計算​

DLT パイプラインで MLflow モデルを使用する​

手動による削除または更新を保持する​