前書き

Databricksにdbtのネイティブアダプターが登場しました。
dbtは国内での知名度はまだまだ高くありませんが、DataOpsの中核コンポーネントとして個人的に期待しているツールです。dbtは例えば、データ加工の流れ（リネージュ）の記録、データモデルのドキュメントの生成、単体テストとは別の観点でデータ品質を自動テストする等、様々な機能をシンプルなSQLとYAMLの設定ファイルで実現することができます。
それのどこが便利なの？というのがなかなか伝わりにくいツールな気もするので、実際に動かしながらdbtの良さを噛みしめていこうと思います。

こちらの記事は、dbt Core integration with Databricksを参考にして、記載しています。

環境

Windows 10 Enterprise (version: 1909)
Visual Studio Code (version: 1.62.3)
Python 3.8.8

DataOpsとは

ガートナー社の発表するハイプサイクルの2018年年版に登場したのがはじまりと言われています。
DevOpsなどのxOpsの流れから生まれた潮流で、私は「データに関する関係部門間の知識・技術・考え方のギャップを埋めて、運用を自動化することによって協力体制を築いていこう」という考え方だと認識しています。
特に、データサイエンティスト・データエンジニア・データオーナー（ビジネスドメインを有している専門家）の３者間ですね。

技術的な要素としては以下のようなもので構成されるものです。

ETL/DWH
データカタログ
データリネージュ
データ品質の自動テスト
データ品質のモニタリング・異常検知
ジョブオーケストレーション/スケジューリング

dbtについて

dbtは、オープンソース版のdbt Coreとクラウドサービス版のdbt Cloudがあります。
dbt Coreは、SQLとYAMLファイルで構成されたdbtプロジェクトとCLIツールを通してDatabricksやsnowflake, bigqueryなどのDWHにクエリしてテーブルやビューを作成します。
dbt Cloudは、dbt Coreに加えて、チーム開発のためにWebベースのIDEやスケジューラー、dbtドキュメントの共有などの機能を提供します。

検証の流れ

環境準備
dbtプロジェクトの作成
dbtのモデル作成と実行　　　　←この記事はここまで
より複雑なモデルの作成と実行
dbtのテスト実行

1. 環境準備

作業ディレクトリの作成

mkdir dbt_demo
cd dbt_demo

Python仮想環境の作成

私は、Pythonのバージョン管理をpyenv、仮想環境をvenvで管理しています。

pyenv local 3.8.8
python -m venv venv
.\venv\Scripts\activate

dbt等のpipインストール

python -m pip install --upgrade pip
pip install dbt-core dbt-databricks

インストールしたライブラリを確認します。

pip list | findstr "dbt databricks"

databricks-sql-connector   0.9.3
dbt                        0.21.1
dbt-bigquery               0.21.1
dbt-core                   0.21.1
dbt-databricks             0.21.1
dbt-extractor              0.4.0
dbt-postgres               0.21.1
dbt-redshift               0.21.1
dbt-snowflake              0.21.1

dbtのバージョンを確認します。

dbt --version

installed version: 1.0.0
   latest version: 1.0.0

Up to date!

Plugins:
  - databricks: 1.0.0

Databricks SQLエンドポイントの作成

dbtから接続するDatabricks SQLエンドポイントを作成します（通常のClusterでもOKです）
SQLエンドポイントの場合は、Connection Detailsに接続情報が記載されています。

またアクセストークンも必要になるので、右下Create a personal access tokenのリンクから
発行しておきます。

テーブルの準備

Databricksノートブックで、diamondsサンプルデータからテーブルを作成しておきます。

csv_file = '/databricks-datasets/Rdatasets/data-001/csv/ggplot2/diamonds.csv'

df = (spark.read
    .format('csv')
    .option("header","true")
    .option("inferSchema", "true")
    .load(csv_file)
)
df.write.saveAsTable("diamonds")

diamondsテーブルはこんなです。

display(df)

2. dbtプロジェクトの作成

プロジェクトというのはdbtの管理単位でSQLとYAMLファイルで構成されます。
dbt initコマンドを実行すると対話的にプロジェクトを作ることができ、
一般的には以下のようなディレクトリ構成で作成されます。

├─ dbt_project.yml
├─ README.md
├─ analyses
├─ data
├─ dbt_packages
├─ logs
├─ macros
├─ models
├─ snapshots
└─ tests

上記ディレクトリやファイルの他にDWHなどへの接続情報は~/.dbt/profiles.ymlに記録されます。profile.ymlは以下のような構造です。

~/.dbt/profiles.yml

databricks_cluster:
  outputs:
    dev:
      host: xxxxx.cloud.databricks.com
      http_path: /sql/1.0/endpoints/8888
      schema: default
      threads: 1
      token: dapiXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX
      type: databricks
  target: dev

dbt initコマンドを実行して、SQLエンドポイントの接続情報やアクセストークンを入力します。

dbt init demo_project

00:00:00  Running with dbt=1.0.0
00:00:00  Creating dbt configuration folder at ~\.dbt
00:00:00  Setting up your profile.
Which database would you like to use?
[1] databricks

(Don't see the one you want? https://docs.getdbt.com/docs/available-adapters)

Enter a number: 1
host (yourorg.databricks.com): xxxxx.cloud.databricks.com
http_path (HTTP Path): /sql/1.0/endpoints/8888
token (dapiXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX):
schema (default schema that dbt will build objects in): default
threads (1 or more) [1]: 1
00:00:00  Profile databricks_cluster written to ~\.dbt\profiles.yml using target's profile_template.yml and your supplied values. Run 'dbt debug' to validate the connection.

dbtからDatabricksのSQLエンドポイントへの接続をテストします。

cd demo_project
dbt debug

00:00:00  Running with dbt=1.0.0
dbt version: 1.0.0
python version: 3.8.8
python path: ...\venv\scripts\python.exe
os info: Windows-10-10.0.xx
Using profiles.yml file at ~\.dbt\profiles.yml
Using dbt_project.yml file at \dbt_demo\demo_project\dbt_project.yml

Configuration:
  profiles.yml file [OK found and valid]
  dbt_project.yml file [OK found and valid]

Required dependencies:
 - git [OK found]

Connection:
  host: xxxxx.cloud.databricks.com
  http_path: /sql/1.0/endpoints/8888
  schema: default
  Connection test: [OK connection ok]

All checks passed!

3. dbtのモデル作成と実行

dbtのモデルはdbtの肝となるデータモデルを扱う概念ですが、その実体は単なるSELECT文で、modelsディレクトリにSQLファイルとして配置します。
モデル自体は単なるSELECT文ですが、dbt_projet.ymlまたはモデルSQLファイルのconfigブロックに、モデル設定を記述することで、dbt runコマンドは良しなに環境ごとにビューやテーブルを作成したり、モデル間の依存関係を記録したりと気の利いた処理をしてくれます。

公式ドキュメント: dbt Models

今回は、SQLファイルにconfigブロックを記述していきます。

dbt_demo/demo_project/models/diamonds_four_cs.sql

{{ config(
  materialized='table',
  file_format='delta'
) }}

select carat, cut, color, clarity
from diamonds

dbt_demo/demo_project/models/diamonds_list_colors.sql

select distinct color
from {{ ref('diamonds_four_cs') }}
sort by color asc

dbt_demo/demo_project/models/diamonds_price.sql

select color, avg(price) as price
from dbt_schema.diamonds
group by color
order by price desc

モデル（SQLファイル）を作成したら、dbt runコマンドで実行します。

dbt run --model models/diamonds_four_cs.sql models/diamonds_list_colors.sql models/diamonds_prices.sql

00:00:00  Running with dbt=1.0.0
00:00:00  Found 5 models, 4 tests, 0 snapshots, 0 analyses, 190 macros, 0 operations, 0 seed files, 0 sources, 0 exposures, 0 metrics
00:00:00
00:00:00  Concurrency: 1 threads (target='dev')
00:00:00
00:00:00  1 of 3 START table model diamonds_four_cs............................ [RUN]
00:00:00  1 of 3 OK created table model diamonds_four_cs....................... [OK in 4.03s]
00:00:00  2 of 3 START view model diamonds_prices.............................. [RUN]
00:00:00  2 of 3 OK created view model diamonds_prices......................... [OK in 1.05s]
00:00:00  3 of 3 START view model diamonds_list_colors......................... [RUN]
00:00:00  3 of 3 OK created view model diamonds_list_colors.................... [OK in 1.37s]
00:00:00
00:00:00  Finished running 1 table model, 2 view models in 10.51s.
00:00:00
00:00:00  Completed successfully
00:00:00
00:00:00  Done. PASS=3 WARN=0 ERROR=0 SKIP=0 TOTAL=3

以上で、dbtからDatabricksにテーブルとビューを作成することができました。

dbt with Databricksで始めるDataOps(2)では、さらにデータ品質のテストの機能を試していきます。