Python を使用してパイプライン コードを開発する

Lakeflow 宣言型パイプラインでは、パイプラインで具体化されたビューとストリーミング テーブルを定義するための新しい Python コードコンストラクトがいくつか導入されています。 パイプライン開発のための Python サポートは、PySpark DataFrame API と Structured Streaming API の基本に基づいています。

Python と DataFrame に慣れていないユーザーの場合は、SQL インターフェイスを使うことをお勧めします。 「SQL でパイプライン コードを開発する」を参照してください。

Lakeflow 宣言パイプライン Python 構文の完全なリファレンスについては、「 Lakeflow 宣言型パイプライン Python 言語リファレンス」を参照してください。

パイプライン開発のための Python の基本

Lakeflow 宣言パイプライン データセットを作成する Python コードは、DataFrames を返す必要があります。

すべての Lakeflow 宣言型パイプライン Python API は、 dlt モジュールに実装されます。 Python で実装されている Lakeflow 宣言型パイプライン コードでは、Python ノートブックとファイルの先頭にある dlt モジュールを明示的にインポートする必要があります。

パイプラインの構成時に指定されたカタログとスキーマに対する既定の読み取りと書き込みを行います。 「ターゲット カタログとスキーマのを設定する」を参照してください。

Lakeflow 宣言型パイプライン固有の Python コードは、1 つの重要な方法で他の種類の Python コードとは異なります。Python パイプライン コードは、データ インジェストと変換を実行して Lakeflow 宣言パイプライン データセットを作成する関数を直接呼び出しません。 代わりに、Lakeflow 宣言パイプラインは、パイプラインで構成されているすべてのソース コード ファイル内の dlt モジュールのデコレーター関数を解釈し、データフロー グラフを構築します。

Python を使用して具体化されたビューまたはストリーミング テーブルを作成する

@dlt.tableデコレーターは、関数によって返された結果に基づいて具体化されたビューまたはストリーミング テーブルを作成するように Lakeflow 宣言パイプラインに指示します。 バッチ読み取りの結果は具体化されたビューを作成し、ストリーミング読み取りの結果はストリーミング テーブルを作成します。

既定では、具体化されたビュー名とストリーミング テーブル名は関数名から推論されます。 次のコード例は、具体化されたビューとストリーミング テーブルを作成するための基本的な構文を示しています。

import dlt

@dlt.table()
def basic_mv():
  return spark.read.table("samples.nyctaxi.trips")

@dlt.table()
def basic_st():
  return spark.readStream.table("samples.nyctaxi.trips")

必要に応じて、name デコレーターの @dlt.table 引数を使用してテーブル名を指定できます。 次の例は、具体化されたビューとストリーミング テーブルのこのパターンを示しています。

import dlt

@dlt.table(name = "trips_mv")
def basic_mv():
  return spark.read.table("samples.nyctaxi.trips")

@dlt.table(name = "trips_st")
def basic_st():
  return spark.readStream.table("samples.nyctaxi.trips")

オブジェクト ストレージからデータを読み込む

Lakeflow 宣言型パイプラインでは、Azure Databricks でサポートされているすべての形式からのデータの読み込みがサポートされています。 「データ形式のオプション」を参照してください。

Databricks では、クラウド オブジェクト ストレージに格納されているデータに対して増分インジェスト ワークロードを構成するときに、自動ローダーとストリーミング テーブルを使用することをお勧めします。 「自動ローダー」を参照してください。

次の例では、自動ローダーを使用して JSON ファイルからストリーミング テーブルを作成します。

import dlt

@dlt.table()
def ingestion_st():
  return (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .load("/databricks-datasets/retail-org/sales_orders")
  )

次の例では、バッチ セマンティクスを使用して JSON ディレクトリを読み取り、具体化されたビューを作成します。

import dlt

@dlt.table()
def batch_mv():
  return spark.read.format("json").load("/databricks-datasets/retail-org/sales_orders")

期待値を使用してデータを検証する

期待値を使用して、データ品質の制約を設定および適用できます。 パイプラインの期待を使用してデータ品質を管理する方法については、を参照してください。

次のコードでは、 @dlt.expect_or_drop を使用して、データ インジェスト中に null レコードを削除する valid_data という名前の期待値を定義します。

import dlt

@dlt.table()
@dlt.expect_or_drop("valid_date", "order_datetime IS NOT NULL AND length(order_datetime) > 0")
def orders_valid():
  return (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .load("/databricks-datasets/retail-org/sales_orders")
  )

パイプラインで定義された具体化されたビューとストリーミング テーブルに対してクエリを実行する

次の例では、4 つのデータセットを定義します。

• JSON データを読み込む orders という名前のストリーミング テーブル。
• CSV データを読み込む customers という名前の具体化されたビュー。
• customer_ordersという名前のマテリアライズドビューは、ordersおよびcustomersデータセットのレコードを結合し、注文タイムスタンプを日付にキャストし、customer_id、order_number、state、およびorder_dateフィールドを選択します。
• 各状態の注文の日次数を集計する daily_orders_by_state という名前の具体化されたビュー。

import dlt
from pyspark.sql.functions import col

@dlt.table()
@dlt.expect_or_drop("valid_date", "order_datetime IS NOT NULL AND length(order_datetime) > 0")
def orders():
  return (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .load("/databricks-datasets/retail-org/sales_orders")
  )

@dlt.table()
def customers():
    return spark.read.format("csv").option("header", True).load("/databricks-datasets/retail-org/customers")

@dlt.table()
def customer_orders():
  return (spark.read.table("orders")
    .join(spark.read.table("customers"), "customer_id")
      .select("customer_id",
        "order_number",
        "state",
        col("order_datetime").cast("int").cast("timestamp").cast("date").alias("order_date"),
      )
  )

@dlt.table()
def daily_orders_by_state():
    return (spark.read.table("customer_orders")
      .groupBy("state", "order_date")
      .count().withColumnRenamed("count", "order_count")
    )

for ループでテーブルを作成する

Python for ループを使用して、複数のテーブルをプログラムで作成できます。 これは、少数のパラメーターによって異なるデータ ソースやターゲット データセットが多数ある場合に便利です。その結果、コードの合計が少なくなり、コードの冗長性が低下します。

for ループはロジックをシリアル順に評価しますが、データセットの計画が完了すると、パイプラインはロジックを並列で実行します。

次の例では、リージョン別に顧客の注文をフィルター処理する 5 つのテーブルを作成します。 ここでは、リージョン名を使用して、ターゲットの具体化されたビューの名前を設定し、ソース データをフィルター処理します。 一時ビューは、最終的な具体化されたビューの構築に使用されるソース テーブルからの結合を定義するために使用されます。

import dlt
from pyspark.sql.functions import collect_list, col

@dlt.view()
def customer_orders():
  orders = spark.read.table("samples.tpch.orders")
  customer = spark.read.table("samples.tpch.customer")

  return (orders.join(customer, orders.o_custkey == customer.c_custkey)
    .select(
      col("c_custkey").alias("custkey"),
      col("c_name").alias("name"),
      col("c_nationkey").alias("nationkey"),
      col("c_phone").alias("phone"),
      col("o_orderkey").alias("orderkey"),
      col("o_orderstatus").alias("orderstatus"),
      col("o_totalprice").alias("totalprice"),
      col("o_orderdate").alias("orderdate"))
  )

@dlt.view()
def nation_region():
  nation = spark.read.table("samples.tpch.nation")
  region = spark.read.table("samples.tpch.region")

  return (nation.join(region, nation.n_regionkey == region.r_regionkey)
    .select(
      col("n_name").alias("nation"),
      col("r_name").alias("region"),
      col("n_nationkey").alias("nationkey")
    )
  )

# Extract region names from region table

region_list = spark.read.table("samples.tpch.region").select(collect_list("r_name")).collect()[0][0]

# Iterate through region names to create new region-specific materialized views

for region in region_list:

  @dlt.table(name=f"{region.lower().replace(' ', '_')}_customer_orders")
  def regional_customer_orders(region_filter=region):

    customer_orders = spark.read.table("customer_orders")
    nation_region = spark.read.table("nation_region")

    return (customer_orders.join(nation_region, customer_orders.nationkey == nation_region.nationkey)
      .select(
        col("custkey"),
        col("name"),
        col("phone"),
        col("nation"),
        col("region"),
        col("orderkey"),
        col("orderstatus"),
        col("totalprice"),
        col("orderdate")
      ).filter(f"region = '{region_filter}'")
    )

このパイプラインのデータ フロー グラフの例を次に示します。

トラブルシューティング: for ループによって、同じ値を持つ多数のテーブルが作成される

パイプラインが Python コードの評価に使用する遅延実行モデルでは、 @dlt.table() によって修飾された関数が呼び出されたときに、ロジックが個々の値を直接参照する必要があります。

次の例では、 for ループを使用してテーブルを定義する 2 つの正しい方法を示します。 どちらの例でも、 tables リストの各テーブル名は、 @dlt.table()によって修飾された関数内で明示的に参照されます。

import dlt

# Create a parent function to set local variables

def create_table(table_name):
  @dlt.table(name=table_name)
  def t():
    return spark.read.table(table_name)

tables = ["t1", "t2", "t3"]
for t_name in tables:
  create_table(t_name)

# Call `@dlt.table()` within a for loop and pass values as variables

tables = ["t1", "t2", "t3"]
for t_name in tables:

  @dlt.table(name=t_name)
  def create_table(table_name=t_name):
    return spark.read.table(table_name)

次の例は 参照を正しくしていません。 この例では、個別の名前を持つテーブルを作成しますが、すべてのテーブルは、 for ループの最後の値からデータを読み込みます。

import dlt

# Don't do this!

tables = ["t1", "t2", "t3"]
for t_name in tables:

  @dlt.table(name=t_name)
  def create_table():
    return spark.read.table(t_name)

具体化されたビューまたはストリーミング テーブルからレコードを完全に削除する

GDPR コンプライアンスの場合など、削除ベクトルが有効になっている具体化されたビューまたはストリーミング テーブルからレコードを完全に削除するには、オブジェクトの基になる Delta テーブルに対して追加の操作を実行する必要があります。 具体化されたビューからレコードを確実に削除するには、「削除ベクターを有効にしてマテリアライズドビューからレコードを完全に削除する」を参照してください。 ストリーミング テーブルからレコードを確実に削除するには、「ストリーミング テーブル からレコードを完全に削除する」を参照してください。