Delta Live Tables パイプラインでパラメーターを使用する

この記事では、Delta Live Tables パイプライン構成を使用してパイプラインコードをパラメーター化する方法について説明します。

Reference パラメーター

更新中、パイプラインソースコードは構文を使用してパイプラインパラメーターにアクセスし、 Spark 構成の値を取得できます。

パイプラインパラメーターは、キーを使用して参照します。値は、ソースコードロジックが評価される前に、文字列としてソースコードに挿入されます。

次の構文例では、キー source_catalog と値 dev_catalog を持つパラメーターを使用して、具体化されたビューのデータソースを指定します。

CREATE OR REFRESH MATERIALIZED VIEW transation_summary AS
SELECT account_id,
  COUNT(txn_id) txn_count,
  SUM(txn_amount) account_revenue
FROM ${source_catalog}.sales.transactions_table
GROUP BY account_id

import dlt
from pyspark.sql.functions import col, sum, count

@dlt.table
def transaction_summary():

  source_catalog = spark.conf.get("source_catalog")

  return (spark.read
      .table(f"{source_catalog}.sales.transactions_table")
      .groupBy("account_id")
      .agg(
        count(col("txn_id").alias("txn_count")),
        sum(col("txn_amount").alias("account_revenue"))
      )
    )

Set パラメーター

任意のキーと値のペアをパイプラインの構成として渡して、パラメーターをパイプラインに渡します。パイプライン設定を定義または編集する際に、ワークスペース UI または JSON を使用してパラメーターを設定できます。「Delta Live Tables パイプラインの構成」を参照してください。

パイプラインパラメーターキーには、 _ - . 文字または英数字のみを含めることができます。パラメーターの値は文字列として設定されます。

パイプラインパラメーターは、動的な値をサポートしていません。パイプライン設定のキーに関連付けられている値を更新する必要があります。

重要

予約済みパイプラインまたは Apache Spark 設定値と競合するキーワードは使用しないでください。

Python または SQL でのデータセット宣言のパラメーター化

データセットを定義する Python コードと SQL コードは、パイプラインの設定によってパラメーター化できます。パラメータ化により、次のユースケースが可能になります。

長いパスとその他の変数をコードから分離します。
開発環境またはステージング環境で処理されるデータ量を減らして、テストを高速化します。
同じ変換ロジックを再利用して、複数のデータソースから処理します。

次の例では、 startDate 設定値を使用して、開発パイプラインを入力データのサブセットに制限します。

CREATE OR REFRESH MATERIALIZED VIEW customer_events
AS SELECT * FROM sourceTable WHERE date > '${mypipeline.startDate}';

@dlt.table
def customer_events():
  start_date = spark.conf.get("mypipeline.startDate")
  return read("sourceTable").where(col("date") > start_date)

{
  "name": "Data Ingest - DEV",
  "configuration": {
    "mypipeline.startDate": "2021-01-02"
  }
}

{
  "name": "Data Ingest - PROD",
  "configuration": {
    "mypipeline.startDate": "2010-01-02"
  }
}

パラメーターを使用した Data の制御

パイプラインパラメーターを使用して、同じパイプラインの異なる構成で異なるデータソースを指定できます。

たとえば、変数 p を使用して、パイプラインの開発、テスト、および本番運用の構成で異なるパスを指定し、次のコードを使用して参照 data_source_path 。

CREATE STREAMING TABLE bronze
AS (
    SELECT
    *,
    _metadata.file_path AS source_file_path
    FROM read_files( '${data_source_path}', 'csv',
            map("header", "true"))
)

import dlt
from pyspark.sql.functions import col

data_source_path = spark.conf.get("data_source_path")

@dlt.table
def bronze():
    return (spark.readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "csv")
        .option("header", True)
        .load(data_source_path )
        .select("*", col("_metadata.file_path").alias("source_file_name"))
    )

このパターンは、インジェストロジックが初期インジェスト中にスキーマまたは不正な形式のデータをどのように処理するかをテストするのに役立ちます。データセットを切り替えながら、すべての環境でパイプライン全体で同じコードを使用できます。