APC 技術ブログ

株式会社エーピーコミュニケーションズの技術ブログです。

株式会社 エーピーコミュニケーションズの技術ブログです。

トップ > #Databricks > Databricksアセットバンドルを使用してLakeflow宣言型パイプラインを管理する

Databricksアセットバンドルを使用してLakeflow宣言型パイプラインを管理する

#Databricks

はじめに

GDAI事業部 Lakehouse部の山下です。最近、Visual Studio Code経由でDatabricksアセットバンドルを利用したLakeflow宣言型パイプラインの開発方法があることを知りました。その内容について、公式チュートリアルを参考にし、実際に試した際の注意点と補足説明を添えた記事を書きたいと思います。

Databricksでの開発手法について

Web UIでの開発が主流だったDatabricksは、今や本格的な開発プラットフォームへと進化しています。この変化の中心にあるのが、Databricks Asset BundlesとTerraformなどのIaC（Infrastructure as Code）ツールです。

Databricks Asset Bundlesは、パイプラインやジョブといったアプリケーションをコードで管理し、CI/CDに組み込むことを可能するAoC（Application as Code）です。また、TerraformなどのIaCツールを使えば、データベースやスキーマなどのインフラリソースもコードで整備できます。

これらのツールをローカルエディターから直接扱うことで、インフラとアプリケーションの両方をコードとして管理する現代的な開発手法をDatabricksに適用する敷居が大幅に下がり、チームでの開発がよりスムーズになります。

事前準備

開発環境の構築のため、各種ツールのインストールを行います。

VsCode のセットアップ

こちらからインストーラーをダウンロードし、インストールしてください。

Databricks CLI のインストール

こちらでインストーラーの案内がされています。インストール後にパスを通して利用可能となります。

バージョン情報が確認できましたら完了です。

Lakeflow宣言型パイプラインの開発

冒頭で触れたように今回は公式チュートリアルを参考に、構築済みのサンプルのLakehouse宣言型パイプラインの開発を行います。

ステップ 1. 認証設定

①次のコマンドを実行します。 任意のプロファイル名を入力することで新規のプロファイルが作成されます。

また、コマンドの実行でWebブラウザにログイン案内の画面が表示されます。

②Webブラウザにログインします。

ログインした際にOAuthトークンが自動的にプロファイルに反映されます。 このコマンド（databricks auth login）の目的は、対話的なWebブラウザのログインフローを通じて、短期的なOAuthトークンを自動で取得し、それをローカルのDatabricks構成プロファイルに安全に保存することです。 ※ユーザーはトークンを手動で生成したり、貼り付けたりする必要がありません。このプロセスは、CLIとWebブラウザ間の連携によって自動的に行われます。

トラブルシューティング

  • 自動での認証の設定がうまくいかなかった場合や、認証方法を切り替えたい場合には別の認証方法として、 databricks configure コマンドでPersonal access tokenを利用した手動の認証設定が可能です。

① Personal access token の取得

まずはDatabricksにログインし、アクセストークンを取得します。

右上のアイコン>設定>左ペインのユーザータブの開発者>開発者ページのアクセストークンの"管理"を選択します。

新規トークンを生成>コメントと存続期間を入力し、"生成"を押下します。

生成したトークンをコピーします。 ※一度生成されたトークンの再表示はできないので注意が必要です。

②手動の認証設定

次のコマンドを実行し、ホストと先ほどコピーしたアクセストークンを設定して保存します。

※こちらで使用したプロファイル名はデフォルト設定になっていない限り、環境に接続するするコマンドを実行する際は -p オプションで指定が必要になるためご注意ください。

ステップ 2: バンドル作成

次のコマンドを実行します。Template to useの場合は、Enterを押してデフォルト値の default-python のままにします。

※今回はテンプレートを利用するため、デフォルト値の default-python のまま進みます。

"Unique name for this project" でプロジェクト名を求められます。今回は”sample_pipeline_project”として作成します。

続いて、プロジェクト構成の設定について選択します。

  • Include a stub (sample) notebook in '...':

noと入力し、Enter キーを押します。 これは、Lakeflowのパイプラインコードを含まない通常のサンプルノートブックを含めないことを意味します。

  • Include a stub (sample) Delta Live Tables pipeline in '...':

Enter キーを押して、デフォルトの yes を受け入れます。 これにより、Lakeflowの宣言型パイプラインコードを含むサンプルノートブックがプロジェクトに追加されます。

  • Include a stub (sample) Python package in '...':

noと入力し、Enter キーを押します。 これは、Pythonのパッケージファイルを含めないことを意味します。

  • Use serverless compute:

yesと入力し、Enter キーを押します。 これにより、作成されるプロジェクトがサーバーレスコンピュートで実行されるように設定されます。

これらの手順を終えると、Databricks CLIは指定した設定に基づいて、databricks.ymlファイルやサンプルコードを含む、新しいプロジェクトフォルダを自動的に生成します。

ステップ 3: バンドルの探索

フォルダ構成の中で、Databricksアセットバンドルに直接関連する主要な設定ファイルは以下です。

.
├── .databricks
├── resources
│   ├── sample_pipeline_project.job.yml
│   └── sample_pipeline_project.pipeline.yml
├── src
│   └── pipeline.ipynb
└── databricks.yml

それぞれのフォルダとファイルの役割は以下の通りです。

.databricks

.databricksフォルダは、Databricksアセットバンドルプロジェクトのルートディレクトリに自動生成される、隠しフォルダです。このフォルダの主な役割は、ローカル開発環境とDatabricksワークスペースを連携させるための設定ファイルを格納することです。

【主な役割】

  • 認証情報の管理

Databricks CLIがワークスペースに接続するための認証情報（トークンなど）を安全に保存します。これにより、コマンドラインからデプロイやジョブの実行を行う際に、毎回認証情報を入力する必要がなくなります。

  • 設定ファイルの格納

databricks.ymlやpyproject.tomlなどの設定ファイルで指定されたローカルパスを、デプロイ時にDatabricksワークスペースのパスにマッピングするための設定を保持します。

  • 状態管理

バンドルのデプロイ履歴や状態情報など、Databricksアセットバンドルのライフサイクルを管理するためのファイルが格納されます。

resources

resourcesフォルダは、Databricksアセットバンドルプロジェクトにおいて、デプロイするすべてのDatabricksリソースの定義ファイルを格納するためのものです。 このフォルダには、通常YAML形式で記述されたファイルが含まれており、これらのファイルはdatabricks.ymlから読み込まれます。

【主な役割】

  • インフラの定義

パイプライン、ジョブ、クラスターポリシー、Unity Catalogの権限などの設定を記述します。

  • 分離と再利用

複数のYAMLファイルに分割することで、プロジェクトの構成を整理し、設定の再利用を容易にします。例えば、ジョブの定義とパイプラインの定義を別のファイルに分けることができます。

  • GitOpsの統合

バンドルの設定をコードとして管理するため、Gitと組み合わせてCI/CDパイプラインを構築する基盤となります。

【配下ファイル】

  • resources/sample_pipeline_project.job.yml
# このファイルは sample_pipeline_project のメインとなるジョブを定義します。
# このジョブは、sample_pipeline_project_pipeline のスケジュールに従って実行されます。
resources:
  # Databricksワークスペースにデプロイされるリソースを定義します。
  jobs:
    # ジョブの名前です。この名前でDatabricks UIに表示されます。
    sample_pipeline_project_job:
      # ジョブの表示名を指定します。
      name: sample_pipeline_project_job

      # このジョブを実行するトリガーを定義します。
      trigger:
        # 最後に実行されてから正確に1日後に、このジョブを毎日実行します。
        # 詳細は https://docs.databricks.com/api/workspace/jobs/create#trigger を参照してください。
        periodic:
          # 実行間隔を数値で指定します。
          interval: 1
          # 実行間隔の単位（DAYS, HOURS, MINUTESなど）を指定します。
          unit: DAYS

      # ジョブの実行結果に関するメール通知設定です。
      # email_notifications:
      #   on_failure:
      #     - your_email@example.com

      # このジョブで実行されるタスクのリストを定義します。
      tasks:
        # タスクの一意なキーです。タスクの識別子として使われます。
        - task_key: refresh_pipeline
          # パイプラインを更新するタスクを定義します。
          pipeline_task:
            # 更新するパイプラインのIDを動的に参照します。
            # `${resources.pipelines.sample_pipeline_project_pipeline.id}`は、
            # 同じバンドル内で定義されたパイプラインのリソースIDを自動で取得します。
            pipeline_id: ${resources.pipelines.sample_pipeline_project_pipeline.id}
  • resources/sample_pipeline_project.pipeline.yml
# sample_pipeline_project のメインパイプラインを定義します。
resources:
  # Databricksワークスペースにデプロイされるパイプラインリソースを定義します。
  pipelines:
    # パイプラインの一意な名前を指定します。
    sample_pipeline_project_pipeline:
      # パイプラインの表示名を指定します。
      name: sample_pipeline_project_pipeline
      # Unity Catalogを使用する場合、カタログ名は必須です。
      catalog: main
      # パイプラインがテーブルを作成するスキーマを定義します。
      # `${bundle.target}`は、デプロイターゲット（例：'dev'または'prod'）に基づいて動的に変わります。
      schema: sample_pipeline_project_${bundle.target}
      # Databricksサーバーレスコンピュートを有効にします。
      # これにより、クラスターを意識することなくパイプラインを実行できます。
      serverless: true
      # パイプラインで実行されるライブラリ（ノートブックやファイルなど）を定義します。
      libraries:
        - notebook:
            # ノートブックのファイルパスを相対パスで指定します。
            path: ../src/pipeline.ipynb

      # パイプラインの実行中に使用される設定値を定義します。
      configuration:
        # パイプラインのソースコードが格納されているディレクトリのパスを指定します。
        # これにより、ソースコード内のファイルに簡単にアクセスできるようになります。
        bundle.sourcePath: ${workspace.file_path}/src

src

srcフォルダは、Databricksアセットバンドルプロジェクトのソースコードを集約して格納するための場所です。こうすることで、プロジェクトの構造が整理され、Gitによるバージョン管理やCI/CDパイプラインへの統合が容易になります。

具体的には、以下のようなファイルが格納されます。

  1. DLTパイプラインのソースコード: @dlt.tableで始まるPythonファイルや、SQLクエリが記述されたファイルなど。

  2. Databricksジョブのノートブック: ジョブで実行するノートブックファイル（.ipynbや.py）。

  3. 共通のライブラリ: 複数のノートブックで利用されるカスタムPythonライブラリなど。

【配下のファイル】

  • src/pipeline.ipynb
import dlt
from pyspark.sql.functions import expr
from pyspark.sql import SparkSession

# Sparkセッションを明示的に取得します。DLTパイプラインでは通常必要ありませんが、
# ノートブックでの開発時に互換性を保つために役立ちます。
spark = SparkSession.builder.getOrCreate()

# @dlt.view デコレータは、ストリーミングビューを定義します。
# ビューは、データが永続化されず、参照されるたびに実行されます。
# ただし、`dlt.table`で参照される場合、DLTによって最適化されます。
@dlt.view
def taxi_raw():
    # Sparkを使ってJSONファイルを読み込みます。
    # この例では、DBFSパスにあるデータセットを読み込んでいます。
    return spark.read.format("json").load("/databricks-datasets/nyctaxi/sample/json/")


# @dlt.table デコレータは、テーブル（またはマテリアライズドビュー）を定義します。
# データは物理的に格納され、増分更新または完全更新が可能です。
@dlt.table
def filtered_taxis():
    # dlt.read()は、同じDLTパイプライン内で定義された他のテーブルやビューを読み込むための
    # DLT専用のメソッドです。ここでは、上記の `taxi_raw` ビューを参照しています。
    # expr()関数は、SQL式を評価してフィルタリングを行います。
    return dlt.read("taxi_raw").filter(expr("fare_amount < 30"))

databricks.yml

アセットバンドルのすべての設定（パイプライン、ジョブ、クラスターなど）がYAML形式で記述されています。

# Databricksアセットバンドルの定義ファイルです。
# ドキュメントは https://docs.databricks.com/dev-tools/bundles/index.html を参照してください。
bundle:
  # バンドルの名前。Databricks UIで表示されるプロジェクト名になります。
  name: sample_pipeline_project
  # バンドルの一意なID。Gitの変更を追跡するために使われます。
  uuid: XXXXXXXXXXXXXXXXXXXXXXXX

# 他のYAMLファイルをこのバンドルに含めます。
# これにより、設定ファイルを複数に分割して管理できます。
include:
  # 'resources/' ディレクトリ直下の全てのYAMLファイルを含めます。
  - resources/*.yml
  # 'resources/' ディレクトリ内のサブディレクトリにある全てのYAMLファイルを含めます。
  - resources/*/*.yml

# ターゲット環境を定義します。開発、本番など、デプロイ先ごとに設定を分けます。
targets:
  # 開発環境のターゲット設定
  dev:
    # デフォルトでは 'mode: development' を使用して、開発用のコピーを作成します。
    # - デプロイされたリソースには '[dev my_user_name]' のようなプレフィックスが付きます。
    # - ジョブのスケジュールやトリガーはデフォルトで一時停止されます。
    # 詳細は https://docs.databricks.com/dev-tools/bundles/deployment-modes.html を参照してください。
    # 開発モードを指定します。これにより、変更のプレビューや安全なデプロイが可能です。
    mode: development
    # このターゲットをデフォルトとして指定します。'databricks bundle deploy'と実行した場合にこれが使われます。
    default: true
    # デプロイ先のDatabricksワークスペースを指定します。
    workspace:
      # ワークスペースのホストURLです。
      host: https://XXXXXXXXXXXXXXXXXXXXXXXXXX

  # 本番環境のターゲット設定
  prod:
    # 本番モードを指定します。プレフィックスは付かず、スケジュールも有効になります。
    mode: production
    # デプロイ先のDatabricksワークスペースを指定します。
    workspace:
      # ワークスペースのホストURLです。
      host: https://XXXXXXXXXXXXXXXXXXXXXXXXXX
      # デプロイ先のルートパスを明示的に指定します。
      # これにより、本番環境では一箇所にリソースをデプロイするようにします。
      root_path: /Workspace/Users/XXXXXXXXXXXXX/.bundle/${bundle.name}/${bundle.target}
    # このターゲットに適用される権限設定です。
    permissions:
      # 権限を付与するユーザー名を指定します。
      - user_name: XXXXXXXXXXXXXXXXXXXXX
        # 付与する権限レベルを指定します。CAN_MANAGEは完全な管理権限です。
        level: CAN_MANAGE

※機密情報は直接バンドルに含めず、Databricks SecretsまたはCI/CD環境変数を利用して管理する方法が推奨されています。

これらのフォルダ構成を理解することで、Databricksアセットバンドルを使った開発の全体像が掴めます。 この体系的なアプローチにより、チームでのプロジェクト管理やCI/CDの導入が格段にスムーズになるでしょう。

ステップ 4: バンドル設定ファイルの検証

次のコマンドを実行し、バンドル構成が有効かどうかを確認します。

> databricks bundle validate -p sample_pipeline_project

以下の出力をもってDatabricksアセットバンドルの設定ファイルであるdatabricks.ymlの構文と整合性の検証が完了です。

ステップ 5: リモートワークスペースへのデプロイ

次のコマンドを実行し、リモートワークスペースへのデプロイを行います。

> databricks bundle deploy -t dev -p sample_pipeline_project

-t devオプションは、デプロイ先のターゲット環境を指定しています。

これはdatabricks.ymlファイル内で定義されたtargets: dev:ブロックのセクションを指します。

devターゲットが指定された場合、Databricks CLIはdatabricks.ymlファイルのdev設定を読み込み、それに従ってリソースをデプロイします。この設定には、ホストURL、デプロイモード、パーミッションなどが含まれます。

これにより、開発環境と本番環境など、異なる設定を持つ複数の環境に、同じバンドルを簡単にデプロイできるようになります。

以下の出力をもってリモートワークスペースへのデプロイが完了です。

※デプロイされたかどうかについては、Databricks ワークスペースのサイドバーの、[ ジョブとパイプライン] から確認が可能です。

ステップ 6: デプロイしたパイプラインの実行

次のコマンドを実行し、デプロイしたパイプラインの実行を行います。

> databricks bundle run -t dev sample_pipeline_project_pipeline -p sample_pipeline_project

以下の出力をもってデプロイしたパイプラインの実行が完了です。

また、Databricks UIからも実行結果が確認できます。

チュートリアルの内容は以上です。ここまで、Databricksアセットバンドルを使い、ローカルのVS CodeでLakehouseの宣言型パイプラインを構築しました。開発からデプロイ、実行結果の確認までの一連の流れが完了し、効率的なデータパイプライン管理について基本的な内容に触れることができたかと思います。

おわりに

本記事では、Databricksアセットバンドルを活用し、ローカルのVS CodeからLakehouseの宣言型パイプラインを管理する手法をチュートリアルベースで解説しました。このチュートリアルを通じて、Databricksでの開発をWeb UIから脱却し、GitOpsの原則に基づいたモダンなワークフローの実現性を一定確認できたのではないかと思います。

アプリケーションとインフラストラクチャの両方をコードとして管理するIaC/AoC（Infrastructure as Code/Application as Code）実現による、チーム開発の効率化と信頼性の向上という大きなメリットのため、今後はこの知見を使ってCI/CDパイプラインの本格的な導入からテスト自動化を目指して、Databricksプロジェクトに活かしていきたいと思います。

私たちはDatabricksを用いたデータ分析基盤の導入から内製化支援等()幅広く支援をしております。もしご興味がある方は、お問い合わせ頂ければ幸いです。

www.ap-com.co.jp www.ap-com.co.jp

また、一緒に働いていただける仲間も募集中です！ エーピーコミュニケーションズやLakehouse部にご興味がある方は問い合わせいただければ幸いです。

hrmos.co

hrmos.co