Delta Live TablesのPython UDFでログを出力する

やってみようとしたら意外に大変だったのでメモ。

[Delta Live Tables(DLT)](https://qiita.com/taka_yayoi/items/0cde1f2732ff859a726f)はDatabricks上で高信頼、高品質なデータパイプラインを作成できるソリューションです。宣言型ETL(最初のテーブルはこういう定義である、次のテーブルの定義は...)を用いることで、処理の手順を記述するのではなく、最終的にどのようなデータが必要なのかを宣言するだけで、途中の処理はDLTがよしなにやってくれます。

パイプラインの定義はPythonかSQLで行います。さらにPythonの[UDF(ユーザー定義関数)](https://qiita.com/taka_yayoi/items/123347aa8323b4a90730)を使用できるので、処理のモジュール化、共通化も行うことができます。

ただ、UDFを作成する際にはデバッグは不可避であり、その際にログ出力ができないと色々不便かと思います。なので、今回ログ出力方法を調査しました。

# 何を使ってログ出力するのか

パッと思いつくのは`print`か`logger`かと思います。結果としてどちらでも行けました。

以下のようにUDFの中でログを出力します。

```py:callee
from pyspark.sql.functions import *
from pyspark.sql.types import *
import math

def log_udf(x) -> int:
    """
    logging udf
    :param x: data to log
    :return: x
    """
    
    import datetime
    import logging
    
    logger = logging.getLogger('test')
    dt_now = datetime.datetime.now()
     
    #raise Exception("デバッグ用メッセージ")
    print(f"{dt_now}: printのデバッグ用メッセージ")
    logger.error(f"{dt_now}: loggerのデバッグ用メッセージ")

    return x

spark.udf.register("log_udf", log_udf, IntegerType())
```

なお、`Exception`を使うとドライバーログ(log4jログ)にログ出力されますが、その度処理が止まるので現実的ではありません。

# どこに出力されるのか

実はこちらで悩みました。DLTの実態はSparkの構造化ストリーミング + Delta Lakeなので、処理もSparkクラスターで実行されます。結論としては、**Spark UIのExecutorタブで表示されるExecutorのstderr**でした。

DLTの画面で**コンピューティング**の**Spark UI**を開きます。**ログ**はドライバーのログなのでこちらを見ても表示されません。
![Screen Shot 2022-06-18 at 14.31.36.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/4d93b09a-da22-4498-c74a-1e092632e623.png)

**Executors**をクリックします。
![Screen Shot 2022-06-18 at 14.32.26.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/52bb981e-880a-588d-0c27-6230cf2bb9c3.png)

下の方にある**Logs**カラムの**stderr**リンクをクリックします。
![Screen Shot 2022-06-18 at 14.33.10.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/633a33f4-eb32-e5b3-418c-a343629bc85d.png)

このようにログを確認することができます。オートスケールするなどして複数のExecutorが存在する場合には、それぞれのログを確認する必要があるので注意してください。
![Screen Shot 2022-06-18 at 14.34.37.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/c726ef2d-4e41-60ce-4d0f-384d82f97d79.png)

# ログを永続化する

DLTのクラスターは停止後は削除されてしまうので、ログも消えてしまいます。これを永続化するには[クラスターログデリバリー](https://qiita.com/taka_yayoi/items/8d951b660cd87c6c5f18#%E3%82%AF%E3%83%A9%E3%82%B9%E3%82%BF%E3%83%BC%E3%83%AD%E3%82%B0%E3%83%87%E3%83%AA%E3%83%90%E3%83%AA%E3%83%BC)を活用して、ログをDBFSに永続化します。

1. Delta Live Tableにアクセスし、対象のパイプラインを開きます。
1. **設定**をクリックします。
1. **パイプライン設定を編集**で**JSON**トグルスイッチをクリックします。
1. `"clusters"`の下に以下の内容を追加します。`"destination"`の値は適宜変更してください。

    ```json:JSON
    "cluster_log_conf": {
                    "dbfs": {
                        "destination": "dbfs:/tmp/takaaki.yayoi@databricks.com/cluster_log"
                    }
    }
    ```
1. 以下のようになっていることを確認して、**保存**をクリックします。
![Screen Shot 2022-06-24 at 9.53.15.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/6ede4ab9-1d43-163b-5d7a-50a580d69e06.png)
1. パイプラインを実行します。
1. ノートブック上でファイルが作成されていることを確認します。`/tmp/takaaki.yayoi@databricks.com/cluster_log/`以降のパスは`クラスターID/executor/アプリケーションID/エグゼキュータノードの連番`という構成になります。
```
%fs
ls /tmp/takaaki.yayoi@databricks.com/cluster_log/0624-004506-zp6uwtgl/executor/app-20220624004529-0000/0
```

![Screen Shot 2022-06-24 at 9.52.55.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/06cf2598-3702-d58f-fd38-5b6c81cd6b5f.png)

こちらを[FileStore](https://qiita.com/taka_yayoi/items/01d9b69d2f5283d27d96)にコピーしてダウンロードして中身を確認することもできます。

```
%fs
cp dbfs:/tmp/takaaki.yayoi@databricks.com/cluster_log/0624-004506-zp6uwtgl/executor/app-20220624004529-0000/0/stderr /FileStore/shared_uploads/takaaki.yayoi@databricks.com
```

![Screen Shot 2022-06-24 at 10.00.13.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/0ca24f5d-f757-cd6d-c054-16471ad68606.png)


### Databricks 無料トライアル

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