Delta Live Tables Pythonリファレンス

[Delta Live Tables Python language reference \| Databricks on AWS](https://docs.databricks.com/data-engineering/delta-live-tables/delta-live-tables-python-ref.html) [2022/3/2時点]の翻訳です。

:::note warn
本書は抄訳であり内容の正確性を保証するものではありません。正確な内容に関しては原文を参照ください。
:::

:::note info
**プレビュー**
本機能は[パブリックプレビュー](https://docs.databricks.com/release-notes/release-types.html)です。サインアップには[Delta Live Tablesへのアクセスをリクエスト](https://databricks.com/p/product-delta-live-tables)してください。
:::

本書ではDelta Live TablesのPythonプログラミングインタフェースの詳細とサンプルを説明します。完全なAPI仕様については[Python API仕様](#python-api仕様)を参照ください。

SQL APIに関しては[Delta Live Tables SQL language reference](https://docs.databricks.com/data-engineering/delta-live-tables/delta-live-tables-sql-ref.html)を参照ください。

# Pythonデータセット

Python APIは`dlt`モジュールで定義されています。Python APIで実装されているDelta Live Tablseパイプラインで`dlt`モジュールをインポートする必要があります。Pythonでビューあるいはテーブルを定義するには関数に対して`@dlt.view`あるいは`@dlt.table`デコレーターを適用します。テーブル名、ビュー名を指定するには関数名あるいは`name`パラメーターを使用できます。以下の例では二つの異なるデータセットを定義しています。入力ソースとしてJSONファイルを受け取る`taxi_raw`というビューと、`taxi_raw`ビューを入力とする`filtered_data`というテーブルです。

```py:Python
@dlt.view
def taxi_raw():
  return spark.read.json("/databricks-datasets/nyctaxi/sample/json/")

# Use the function name as the table name
@dlt.table
def filtered_data():
  return dlt.read("taxi_raw").where(...)

# Use the name parameter as the table name
@dlt.table(
  name="filtered_data")
def create_filtered_data():
  return dlt.read("taxi_raw").where(...)
```

ビューとテーブルは、Spark[データフレーム](https://docs.databricks.com/spark/latest/dataframes-datasets/introduction-to-dataframes-python.html)かKoalas[データフレーム](https://koalas.readthedocs.io/en/latest/reference/api/databricks.koalas.DataFrame.html)を返却する必要があります。関数によって返却されるKoalasデータフレームは、Delta Live TablesランタイムによってSparkデータセットに変換されます。

[外部データソース](https://docs.databricks.com/data-engineering/delta-live-tables/delta-live-tables-data-sources.html)からの読み込みに加え、Delta Live Tablseの`read()`関数を用いて、同じパイプラインで定義されたデータセットにアクセスすることができます。以下の例では、`read()`関数による`customers_filtered`の作成をデモします。

```py:Python
@dlt.table
def customers_raw():
  return spark.read.csv("/data/customers.csv")

@dlt.table
def customers_filteredA():
  return dlt.read("customers_raw").where(...)
```

同じパイプラインで定義されたデータセット、あるいは、メタストアに登録されたテーブルにアクセスするために`spark.table()`を使用することもできます。パイプラインに定義されているデータセットにアクセスするために`spark.table()`を用いる際、関数の引数において、データセット名の先頭に`LIVE`を追加します。

```py:Python
@dlt.table
def customers_raw():
  return spark.read.csv("/data/customers.csv")

@dlt.table
def customers_filteredB():
  return spark.table("LIVE.customers_raw").where(...)
```

メタストアに登録されているテーブルからデータを読み込むには、関数の引数から`LIVE`キーワードを削除し、データベース名(オプション)を伴うテーブル名を指定します。

```py:Python
@dlt.table
def customers():
  return spark.table("sales.customers").where(...)
```

Delta Live Tablesは、パイプラインが自動でデータセット間の依存関係を捕捉することを保証します。この依存関係情報は、アップデートを実行する際の実行順序を決定し、パイプラインのイベントログにリネージュ情報を記録するために用いられます。

クエリー関数で`spark.sql`表現を用いてデータセットを返却することができます。内部データセットから読み込みを行うには、データセット名に`LIVE.`を追加します。

```py:Python
@dlt.table
def chicago_customers():
  return spark.sql("SELECT * FROM LIVE.customers_cleaned WHERE city = 'Chicago'")
```

ビューとテーブルの両方には、以下のオプションのプロパティが存在します。

- `comment`: 人間が理解できるデータセットの説明。
- `spark_conf`: 当該クエリーの実行にのみ適用されるSpark設定を含むPythonのディクショナリー。
- [エクスペクテーション](https://docs.databricks.com/data-engineering/delta-live-tables/delta-live-tables-expectations.html)によって強制されるデータ品質制約。

テーブルでは、自身のマテリアライゼーションに対する追加の制御手段を提供しています。

- `partition_cols`を用いて、どのようにテーブルの[パーティションを作成](https://docs.delta.io/latest/best-practices.html#choose-the-right-partition-column)するのかを指定します。クエリーの性能を改善するためにパーティショニングを活用することができます。
- ビュー、テーブルを定義する際にテーブルプロパティを設定することができます。詳細は[テーブルプロパティ](#テーブルプロパティ)を参照ください。
- `path`セッティングを用いることでテーブルデータの格納場所を設定します。デフォルトでは、`path`が設定されていない場合、テーブルデータはパイプラインの格納場所に保存されます。
- Pythonの`StructType`あるいはSQL DDL文字列を用いて、テーブルスキーマをオプションとして指定することができます。以下の例では、明示的に指定したスキーマを用いて`sales`というテーブルを作成します。

```py:Python
sales_schema = StructType([
  StructField("customer_id", StringType(), True),
  StructField("number_of_line_items", StringType(), True),
  StructField("order_datetime", StringType(), True),
  StructField("order_number", LongType(), True)]
)

@dlt.table(
  comment="Raw data on sales",
  schema=sales_schema)
def sales():
  return ("...")

@dlt.table(
  comment="Raw data on sales",
  schema="customer_id STRING, customer_name STRING, number_of_line_items STRING, order_datetime STRING, order_number LONG")
def sales():
  return ("...")
```

デフォルトでは、Delta Live Tablesはスキーマを指定しない場合、`table`定義からスキーマを推定します。

# Pythonライブラリ

外部のPythonライブラリを指定するには、`%pip install`マジックコマンドを使用します。アップデートがスタートすると、Delta Live Tablesはいかなるテーブル定義を実行する前に、`%pip install`コマンドを含む全てのセルを実行します。パイプラインに含まれる全てのPythonノートブックは、インストールされた全てのライブラリにアクセスできます。以下の例では、パイプラインの全てのPythonノートブックから利用できる`logger`というパッケージをインストールしています。

```py:Python
%pip install logger

from logger import log_info

@dlt.table
def dataset():
    log_info(...)
    return dlt.read(..)
```

# Python API仕様

## Pythonモジュール

Delta Live TablesのPython関数は`dlt`モジュールで定義されています。Python APIを用いて実装されたパイプラインでは、このモジュールをインポートする必要があります。

```py:Python
import dlt
```

## テーブルの作成

Pythonでテーブルを定義するには、`@table`デコレーターを適用します。`@table`デコレーターは`@create_table`デコレーターのエイリアスです。

```py:Python
import dlt

@dlt.table(
  name="<name>",
  comment="<comment>",
  spark_conf={"<key>" : "<value", "<key" : "<value>"},
  table_properties={"<key>" : "<value>", "<key>" : "<value>"},
  path="<storage-location-path>",
  partition_cols=["<partition-column>", "<partition-column>"],
  schema="schema-definition",
  temporary=False)
@dlt.expect
@dlt.expect_or_fail
@dlt.expect_or_drop
@dlt.expect_all
@dlt.expect_all_or_drop
@dlt.expect_all_or_fail
def <function-name>():
    return (<query>)
```

## ビューの作成

Pythonでビューを定義するには、`@view`デコレーターは`@create_view`デコレーターのエイリアスです。

```py:Python
import dlt

@dlt.view(
  name="<name>",
  comment="<comment>")
@dlt.expect
@dlt.expect_or_fail
@dlt.expect_or_drop
@dlt.expect_all
@dlt.expect_all_or_drop
@dlt.expect_all_or_fail
def <function-name>():
    return (<query>)
```

## Pythonプロパティ

| `@table`あるいは`@view` |
|:--|
|**name**<br>Type: `str`<br>テーブルかビューに対する名前(オプション)。指定されていない場合、関数名がテーブル、ビュー名として用いられます。   |
|**comment**<br>Type: `str`<br>テーブルの説明(オプション)。   |
|**spark_conf**<br>Type: `dict`<br>このクエリーの実行に対するSpark設定のリスト(オプション)。   |
|**table_properties**<br>Type: `dict`<br>テーブルに対するテーブルプロパティのリスト(オプション)。   |
|**path**<br>Type: `str`<br>テーブルデータの格納場所(オプション)。設定されない場合、システムはパイプラインの格納場所を使用します。   |
|**partition_cols**<br>Type: `array`<br>テーブルのパーティションに使用する1つ以上のカラムのリスト(オプション)。   |
|**schema**<br>Type: `str`か`StructType`<br>テーブルに対するスキーマ定義(オプション)。SQL DDL文字列、Pythonの`StructType`として定義されるスキーマ。   |
|**temporary**<br>Type: `bool`<br>一時テーブルの作成。このテーブルに対するメタデータは永続化されません。   |

| テーブル、ビュー定義 |
|:--|
|**def <function-name>()**<br>データセットを定義するPython関数。`name`パラメーターが設定されていない場合、`<function-name>`がターゲットのデータセット名となります。   |
|**query**<br>Sparkデータセット、あるいはKoalasデータフレームを返却するSpark SQL文。<br><br>同じパイプラインで定義されるデータセットに対するcomplete読み込みを行うには、`dlt.read()`か`spark.table()`を使用します。同じパイプラインで定義されているデータセットから読み込む際には、spark.table()関数の引数のデータセット名の先頭に`LIVE`キーワードを追加します。例えば、`customers`というデータセットから読み込む際には、<br><br>`spark.table("LIVE.customers")`<br><br>また、メタストアに登録されているテーブルから読み込むケースで`spark.table()`関数を使う際には、`LIVE`キーワードを削除して、データベース名(オプション)を伴うテーブル名を指定します。<br><br>`spark.table("sales.customers")`<br><br>同じパイプラインで定義されているデータセットからインクリメンタルにデータセットを読み込む際は、`dlt.read_stream()`を使用します。<br><br>返却するデータセットを作成するためにSQLクエリーを定義するには`spark.sql`関数を使用します。<br><br>PythonによるDelta Live Tablesクエリーを定義するには[PySpark](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql.html#dataframe-apis)文法を使用します。   |

| エクスペクテーション |
|:--|
|**`@expect(“description”, “constraint”)`**<br><br>`description`で識別されるデータ品質制約を宣言します。レコードがエクスペクテーションに違反した場合、ターゲットデータセットに当該レコードを含めます。|
|**`@expect_or_drop(“description”, “constraint”)`**<br><br>`description`で識別されるデータ品質制約を宣言します。レコードがエクスペクテーションに違反した場合、ターゲットデータセットから当該レコードを削除します。|
|**`@expect_or_fail(“description”, “constraint”)`**<br><br>`description`で識別されるデータ品質制約を宣言します。レコードがエクスペクテーションに違反した場合、即座に処理を停止します。|
|**`@expect_all(expectations)`**<br><br>1つ以上のデータ品質制約を宣言します。`expectations`はPythonディクショナリーであり、キーがエクスペクテーションの名称となり、値がエクスペクテーションの制約となります。いずれかのエクスペクテーションに違反したレコードはターゲットデータセットに含まれます。|
|**`@expect_all_or_drop(expectations)`**<br><br>1つ以上のデータ品質制約を宣言します。`expectations`はPythonディクショナリーであり、キーがエクスペクテーションの名称となり、値がエクスペクテーションの制約となります。いずれかのエクスペクテーションに違反したレコードはターゲットデータセットから削除されます。|
|**`@expect_all_or_fail(expectations)`**<br><br>1つ以上のデータ品質制約を宣言します。`expectations`はPythonディクショナリーであり、キーがエクスペクテーションの名称となり、値がエクスペクテーションの制約となります。いずれかのエクスペクテーションにレコードが違反した場合、即座に処理を停止します。|

# テーブルプロパティ

[Delta Lake](https://qiita.com/taka_yayoi/items/47b58eb8696b278a3a82#%E3%83%86%E3%83%BC%E3%83%96%E3%83%AB%E3%83%97%E3%83%AD%E3%83%91%E3%83%86%E3%82%A3)でサポートされているテーブルプロパティに加え、以下のテーブルプロパティを設定することができます。

| テーブルプロパティ |
|:--|
|**pipelines.autoOptimize.managed**<br>デフォルト値: `true`<br>当該デーブルに対する自動スケジュール最適化を有効化するか、無効化するかを指定します。   |
|**pipelines.autoOptimize.zOrderCols**<br>デフォルト値: None<br>当該テーブルでz-orderを行うカラムをカンマ区切りで指定します(オプション)。    |
|**pipelines.reset.allowed**<br>デフォルト値: `true`<br>当該テーブルに対するフルリフレッシュを許可するかどうかを制御します。    |
|**pipelines.trigger.interval**<br>デフォルト値: フロータイプに依存します。<br>当該テーブルのアップデートを行うフローのトリガー周期を制御します。デフォルト値は以下のようになります。<ul><li>ストリーミングクエリーは4秒<li>全ての入力データがDeltaソースであり、completeクエリーの場合は1分<li>非Deltaのデータソースが含まれており、completeクエリーの場合は10分。[Complete tables in continuous pipelines](https://docs.databricks.com/data-engineering/delta-live-tables/delta-live-tables-concepts.html#complete-tables)をご覧ください。</ul>値は数値と時間ユニットになります。以下に適切な時間ユニットを示します。<ul><li>`second`, `seconds`<li>`minute`, `minutes`<li>`hour`, `hours`<li>`day`, `days`</ul>値を定義する際、以下のように、単数形、複数形のユニットを指定することができます。<ul><li>`{"pipelines.trigger.interval" : "1 hour"}`<li>`{"pipelines.trigger.interval" : "10 seconds"}`<li>`{"pipelines.trigger.interval" : "30 second"}`<li>`{"pipelines.trigger.interval" : "1 minute"}`<li>`{"pipelines.trigger.interval" : "10 minutes"}`<li>`{"pipelines.trigger.interval" : "10 minute"}`</ul>    |

### Databricks 無料トライアル

[Databricks 無料トライアル](https://databricks.com/jp/try-databricks)