dbt on databricks～Databricks上でdbt Coreを実行してみた～

## 概要
dbt with databricksではなく、dbt on databricks、つまり、Databricks上でdbt Coreを実行する方法を検証しましたので内容を共有します。本手順では、Databricksからdbtを経由してsnowflakeへ接続する手順を紹介します。

Databricksは、大量データの書き込みに最適なSpark コネクターを利用できることや静的パブリックIPアドレスを付与できることから、他データストアへのデータ連携のELTにおけるEL（データ書き込み）を実施するツールとして最適です。データ連携のELTにおけるT（データストア内でのデータ変換）のツールとして最適なdbtをdatabricksで実行できれば、最強のツールになるのではないかと考えて検証しました。

https://qiita.com/manabian/items/9d4c43b312095c44e503

Databricks Repos上にdbtのコードを実行します。ただし、dbtのデバッグ時にのみ、下記仕様により、クラスターのストレージにコードをコピーしてから実行する必要があります。

- dbt debugを実行する場合、あるいは、--debugをつけて実行する場合には、dbtのコードと同一ディレクトリにある`logs/dbt.log`にログを書き込むというdbtの仕様があること
- Files in ReposへPythonの関数から参照する場合には読み取りのみに制限されているため、デバッグ時にログファイル（logs/dbt.log）へ書き込むことができないこと

dbtから書き込むディレクトリの設定には下記のものがあり、`log-path`のみdbfs上のディレクトリを指定するとエラーとなることに注意してください。`log-path`のディレクトリを、クラスターのストレージ（例：`/tmp/dbt_logs`）にて記載後、ログファイルを移動する必要があります。

- [target-path](https://docs.getdbt.com/reference/project-configs/target-path)
- [packages-install-path](https://docs.getdbt.com/reference/project-configs/packages-install-path)
- [log-path](https://docs.getdbt.com/reference/project-configs/log-path)

```dbt_project.yml
target-path: '/dbfs/dbt_manabian/target'
packages-install-path: '/dbfs/dbt_manabian/dbt_packages'
log-path: '/tmp/dbt_logs'
```

dbtがどういったツールであるかを理解するためには、下記の記事を参考にしてください。

https://qiita.com/manabian/items/67af7e4476d436aded77

https://qiita.com/manabian/items/ab6c2a5dde78c432ecf1

## 環境準備
### Databricksクラスターにてdbfs上に配置してinitスクリプトを実行するように設定

```init.sh
sudo apt-get libpq-dev python-dev python3-pip
sudo apt-get remove python-cffi
sudo pip install --upgrade cffi

```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/9fb569e2-42c6-ce8d-e991-0d4cd6c449b0.png)

DatabricksはUbuntuで動作していることから、下記の記事を参考にしております。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/a518823a-5bb0-9fc5-7d11-c5a736813edb.png)
引用元：[Use pip to install dbt | dbt Docs (getdbt.com)](https://docs.getdbt.com/dbt-cli/install/pip#ubuntudebian)


### Databricksクラスターにて接続先のデータストアに応じたdbtのライブラリ（本手順ではdbt-snowflake==1.0.0）とcryptography（cryptography~=3.4）をインストール

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/8ed3a38b-9773-7023-0c7a-5c1ecb044fc0.png)

### dbt Coreがインストールされていることを確認

```python
%sh
dbt --version
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/e76f80f9-fdb3-1d1d-8683-1ee7c67abe07.png)

### dbtのコードをDatabricks Repos上に配置

```python
%sh
cd /Workspace/Repos/dbt_test/dbt_test
pwd;find . | sort | sed '1d;s/^\.//;s/\/\([^/]*\)$/|--\1/;s/\/[^/|]*/|  /g'
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/d59d0fa0-ed61-5628-1185-f552a8d96c38.png)

## dbtのコードをデバック
### クラスターのストレージへdbtのコードを配置

```python
# クラスターのストレージへdbt関連のコードを配置
in_dir_path  = 'file:/Workspace/Repos/dbt_test/dbt_test/dbt'
out_dir_path = 'file:/dbt_manabian/dbt_test'

dbutils.fs.rm(out_dir_path, True)
dbutils.fs.cp(in_dir_path, out_dir_path, recurse=True)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/897c7edd-46ea-4f41-480f-7406cf9dbc58.png)

### クラスターのストレージ上のdbtのコードをデバッグ

```python
%sh
export DBT_PROFILES_DIR=profiles/snowflake
cd /dbt_manabian/dbt_test
dbt debug
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/59f80850-4409-f0bf-f038-5f53a24630da.png)


### デバッグ時のログファイルを確認

```python
import pprint
pprint.pprint(dbutils.fs.head('file:/dbt_manabian/dbt_test/logs/dbt.log'))
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/61edf22c-184d-136c-9282-5c2b9e46ab06.png)

## dbtのコードを実行
### dbtのコードを実行

```python
%sh
export DBT_PROFILES_DIR=profiles/snowflake
cd /Workspace/Repos/dbt_test/dbt_test/dbt
dbt deps
dbt docs generate
dbt run
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/48c3d7e4-4246-d422-6e60-881135169021.png)

### ログファイルをdbfsにコピー

```python
# ログファイルをdbfsにコピー 
from datetime import datetime

# 現在の時分秒のUNIX時刻の文字列を取得
now_ts = str(int(datetime.now().timestamp()))

dbutils.fs.cp('file:/tmp/dbt_logs', f'dbfs:/dbt_manabian/dbt_log/{now_ts}', True)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/328dd95e-1820-bd69-e010-3f3b3a943cbd.png)

### dbt関連のファイルが生成されていることを確認

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/d8f74e9b-fd3f-954c-b0d9-1279d09525a7.png)

```python
display(spark.read.json('/dbt_manabian/dbt_log/*/*'))
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/d8fb5fbc-7f7f-aa47-c122-e496406a49c6.png)



## Databricks on dbtを利用する際の制約事象
### Files in ReposへPythonの関数から参照する場合には読み取りのみに制限されているため、ログ等のファイルを書き込むことができない。
Daabricks Repos上にあるdbtのコードを実行すると`logs`配下にファイルを書き込む際にエラーとなります。

```python
%sh
export DBT_PROFILES_DIR=profiles/snowflake
cd /Workspace/Repos/dbt_test/dbt_test
dbt debug
```

> 05:33:00  Encountered an error:
[Errno 38] Function not implemented: 'logs'

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c5c6516f-3264-bb72-afdf-b48f13335e6e.png)


### dbfs上にdbt.logを書き込むと後続のdbt処理がエラーとなる
下記のようにdebugを2回実行すると、2回目の処理がエラーとなります。

```python
%sh
export DBT_PROFILES_DIR=profiles/snowflake
cd /dbfs/dbt_manabian/dbt_test
dbt debug
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/5d55fe1c-187d-1756-abaa-44cb384c6f7e.png)

```python
%sh
export DBT_PROFILES_DIR=profiles/snowflake
cd /dbfs/dbt_manabian/dbt_test
dbt debug
```

> --- Logging error ---
Traceback (most recent call last):
  File "/usr/lib/python3.8/logging/__init__.py", line 1089, in emit
    self.flush()
  File "/usr/lib/python3.8/logging/__init__.py", line 1069, in flush
    self.stream.flush()
OSError: [Errno 95] Operation not supported

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/7b09e771-2ce9-396a-c3f5-a1e92f9aebc2.png)

## 参考記事

- [tree コマンドが無い環境で tree コマンドを実現 - Qiita](https://qiita.com/yone098@github/items/bba8a42de6b06e40983b)