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

[アーティクル]
11/16/2024

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

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

Delta Live Tables Python 構文の完全なリファレンスについては、「 Delta Live Tables Python 言語リファレンスを参照してください。

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

Delta Live Tables データセットを作成する Python コードでは、DataFrame を返す必要があります。

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

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

重要

パイプラインの実行時に予期しない動作が発生しないよう、データセットを定義する関数に、副作用がある可能性のあるコードを含めないでください。詳しくは、Python のリファレンスに関する記事をご覧ください。

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

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

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

Note

どちらの関数も、 samples カタログ内の同じテーブルを参照し、同じデコレーター関数を使用します。これらの例では、具体化されたビューとストリーミングテーブルの基本的な構文の唯一の違いは、 spark.read と spark.readStreamを使用していることです。

すべてのデータソースがストリーミング読み取りをサポートしているわけではありません。一部のデータソースは、常にストリーミングセマンティクスで処理する必要があります。

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")

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

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")

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

Delta Live Tables は、Azure Databricks でサポートされているすべての形式からのデータの読み込みをサポートしています。「データ形式のオプション」を参照してください。

Note

これらの例では、ワークスペースに自動的にマウントされた /databricks-datasets で使用可能なデータを使用します。 Databricks では、クラウドオブジェクトストレージに格納されているデータを参照するために、ボリュームパスまたはクラウド URI を使用することをお勧めします。「Unity Catalog ボリュームとは」を参照してください。

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")

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

期待値を使用して、データ品質の制約を設定および適用できます。「Delta Live Tables を使用してデータ品質を管理する」を参照してください。

次のコードでは、 @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")
  )

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

LIVE スキーマを使用して、パイプラインで定義されている他の具体化されたビューとストリーミングテーブルに対してクエリを実行します。

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

JSON データを読み込む orders という名前のストリーミングテーブル。
CSV データを読み込む customers という名前の具体化されたビュー。
ordersおよびcustomers データセットのレコードを結合し、注文タイムスタンプを日付にキャストし、customer_id、order_number、state、およびorder_dateフィールドを選択する、customer_ordersという名前の具体化されたビュー。
各状態の注文の日次数を集計する 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("LIVE.orders")
    .join(spark.read.table("LIVE.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("LIVE.customer_orders")
      .groupBy("state", "order_date")
      .count().withColumnRenamed("count", "order_count")
    )

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

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

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("LIVE.customer_orders")
    nation_region = spark.read.table("LIVE.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}'")
    )

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

5 つのリージョンテーブルに至る 2 つのビューのデータフローグラフ。

トラブルシューティング: `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)

次の方法で共有

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

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

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

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

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

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

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

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

フィードバック

その他のリソース

次の方法で共有

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

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

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

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

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

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

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

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

フィードバック

その他のリソース

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

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

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

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

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

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