【Databricks】【初心者】DLTを試してみた

# 背景・目的
以前の[【Databricks】【初心者】DLTのチュートリアルを試してみた](https://qiita.com/zumax/items/a7ece4fcd330fce99541)の続きで、Delta Live Tables（以降、DLTという。）について試してみます。

# まとめ
- 上位のパイプラインからデータセットにアクセスするには下記の方法がある
    - dlt.read
    - spark.table
    - spark.sql

# 概要
[Delta Live Tables Python language reference](https://docs.databricks.com/en/delta-live-tables/python-ref.html)をもとに整理します。

## 制限事項
> Delta Live Tables Pythonインターフェイスには次の制限があります：
> - Python table関数とview関数はDataFrameを返す必要があります。DataFrameを操作する一部の関数はDataFrameを返さないため、使用しないでください。DataFrame変換は完全なデータフローグラフが解決された後に実行されるため、このような操作を使用すると、意図しない副作用が発生する可能性があります。これらの操作には、collect()、count()、toPandas()、save()、saveAsTable()などの関数が含まれます。ただし、このコードはグラフの初期化フェーズ中に1回実行されるため、これらの関数をtableまたはview関数定義の外に含めることができます。
> - pivot()関数はサポートされていません。Sparkのpivotオペレーションでは、出力のスキーマを計算するために入力データの積極的な読み込みが必要です。この機能は、Delta Live Tablesではサポートされていません。


- table関数、view関数はDataFrameを返す
- DataFrameのうち、一部の関数はDataFrameを返さないので使用しない
    - Transformは遅延評価により、想定外の動作になる可能性がある
- pivot関数はサポートされてない

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

- dlgのPython関数は、dltモジュールで定義されている
- このモジュールをインポートする必要がある

```
import dlt
```

## Delta Live Tablesマテリアライズドビューまたはストリーミングテーブルを作成する
> Pythonでは、Delta Live Tablesは、定義クエリーに基づいてデータセットをマテリアライズドビューとして更新するかストリーミングテーブルとして更新するかを決定します。@tableデコレータは、マテリアライズドビューとストリーミングテーブルの両方を定義するために使用されます。

- DLTは、定義したクエリに基づきMviewかストリーミングテーブルとして更新するかを決定する
- @tableデコレーダーは、Mviewとストリーミングテーブルの両方を定義するために使用される

> Pythonでマテリアライズドビューを定義するには、データソースに対して静的読み取りを実行するクエリーに@tableを適用します。ストリーミングテーブルを定義するには、データソースに対してストリーミング読み取りを実行するクエリーに@tableを適用します。どちらのデータセットタイプも、次のように同じ構文仕様を持っています：

- Mviewを定義するにはデータソースに対して静的読み取りを実行するクエリに@tableを適用する
- ストリーミングを定義するには、データソースに対してストリーミング読み取りを実行するクエリに@tableを適用する

```
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でテーブルまたはビューを定義するには、@dlt.viewまたは@dlt.tableデコレータを関数に適用します。関数名またはnameパラメーターを使用して、テーブルまたはビューの名前を割り当てることができます。次の例では、2つの異なるデータセットを定義しています。JSONファイルを入力ソースとして受け取るtaxi_rawというビューと、入力としてtaxi_rawビューを受け取るfiltered_dataというテーブルです：

- @dlt.viewまたは@dlt.tableデコレータを関数に適用しテーブルやビューを定義する

```
import dlt

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

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

# Use the name parameter as the table name
@dlt.table(
  name="filtered_data")
def create_filtered_data():
  return dlt.read("taxi_raw")
```
<img src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/02969fda-0318-9c0d-3f7b-48c4a2b6cf43.png" width=400 />




## 例：同じパイプラインで定義されたデータセットにアクセスする
> 外部データソースからの読み取りに加えて、Delta Live Tables read()関数を使用して、同じパイプラインで定義されているデータセットにアクセスできます。次の例は、read()関数を使用してcustomers_filteredデータセットを作成する方法を示しています：
- 外部データソースから読み取りに加えてDLTのread関数を使用して、同じパイプラインで定義されているデータセットにアクセス可能


1. まずは上記と同じdlt.readで試します。
    ```
    @dlt.table
    def nyctaxi_raw():
      return spark.read.format("json").load("/databricks-datasets/nyctaxi/sample/json/")
    
    @dlt.table
    def dlt_read():
      return dlt.read("customers_raw")
    ```
    <img src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/b72c355d-acd2-b088-9c08-341c47d33a93.png" width=400 />

    > spark.table()関数を使用して、同じパイプラインで定義されたデータセットにアクセスすることもできます。spark.table()関数を使用してパイプラインで定義されたデータセットにアクセスするときは、LIVE関数の引数でデータセット名の前にキーワードを追加します：

    - sparkのtable関数を使用することで上記と同じことができる。
    - LIVE関数の引数にデータセットキーワードを指定する


1. 次にspark.table関数により読み込みます。
    ```
    @dlt.table
    def nyctaxi_raw2():
      return spark.read.format("json").load("/databricks-datasets/nyctaxi/sample/json/")
    
    @dlt.table
    def nyctaxi_raw3():
      return spark.table("LIVE.nyctaxi_raw2")
    ```
    <img src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/c22d3163-4450-2362-5acb-0d23a6bee98c.png" width=400 />


## 例：Unity Catalog テーブルからのバッチ インジェスト
> Unity Catalog を使用するように構成されたパイプラインは、以下からデータを読み取ることができます。
> - Unity Catalog マネージドテーブルと外部テーブル、ビュー、具体化されたビュー、ストリーミングテーブルがあります。
> - テーブルとビューHive metastore 。
> - Auto Loader cloud_files() 機能を使用して Unity Catalog 外部から読み取ります。
> - Apache Kafka と Amazon Kinesis。

1. 事前にUnityCatalogで作成したテーブルを確認します。

    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/8ae6a69c-4606-52dd-eb14-4423340082cf.png)

1. 下記のコードを実行します。読み込めました。
    ```
    import dlt
    
    @dlt.table
    def table_name():
      df  = spark.table("test.retail.order_details")
      df.printSchema()
      df.show()
      return df
    ```
    <img src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/4457fcad-bec9-3ec7-35bd-e13067142b8c.png" width=400 />

## 例：spark.sqlを使ってデータセットにアクセスする
> クエリー関数でspark.sql式を使用してデータセットを返すこともできます。内部データセットから読み取るには、データセット名の前にLIVE.を追加します：

- spark.sqlを使って読み込むこともできる
- 内部データセットから読み込むときにLIVEを使用する

    ```
    @dlt.table
    def nyctaxi_sql():
      return spark.sql("SELECT * FROM LIVE.nyctaxi_raw3")
    ```
    <img src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/ef0746a4-fd7c-6dbb-5310-c84b11e36865.png" width =400 />

## 例: 1 回限りのデータ バックフィルを実行する
> 次の例では、クエリーを実行して、ヒストリカルデータをストリーミングテーブルに追加します。
> 注
> バックフィルクエリーがスケジュールされたベースまたは継続的に実行されるパイプラインの一部である場合に、真の 1 回限りのバックフィルを確保するには、パイプラインを一度実行した後にクエリーを削除します。 新しいデータがバックフィル ディレクトリに到着した場合に追加するには、クエリーをそのままにしておきます。

# 考察
今回は、ドキュメントに基づいてDelta Live Tables（DLT）について試してみました。次回はパイプラインの実行を試してみます。
# 参考

https://qiita.com/zumax/items/1d7f300eac4361b767fd

https://qiita.com/zumax/items/a7ece4fcd330fce99541

https://docs.databricks.com/ja/delta-live-tables/python-ref.html#example-define-tables-and-views

https://docs.databricks.com/ja/delta-live-tables/unity-catalog.html#ingest-data-into-a-unity-catalog-pipeline&language-python