Pythonを使用したパイプラインコードの開発

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

PythonやDataFramesに詳しくないユーザーには、SQL インターフェイスを使用することをお勧めします。SQL を使用したパイプラインコードの開発を参照してください。

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

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

PythonDelta Live Tablesデータセットを作成するコードDataFrames 、を返す必要があります。

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

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

重要

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

Pythonを使用したマテリアライズドビューやストリーミングテーブルの作成

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

デフォルトでは、マテリアライズド・ビューとストリーミング・テーブルの名前は関数名から推論されます。次のコード例は、マテリアライズドビューとストリーミングテーブルを作成するための基本的な構文を示しています。

注：

どちらの関数も、 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 は、Databricks でサポートされているすべての形式からのデータの読み込みをサポートしています。データ形式のオプションを参照してください。

注：

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

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

次の例では、Auto Loaderを使用して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}'")
    )

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

トラブルシューティング: ループ`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を使用したパイプライン コードの開発