AWS S3×databricks×dbtでデータパイプラインを構築する方法

## はじめに

データの分散管理が増加する中、AWS S3の柔軟なストレージとDatabricksの分析プラットフォーム、さらにdbt（data build tool）を組み合わせることで、スケーラブルで管理しやすいデータパイプラインを構築できます。

本記事では、以下の内容を段階的に解説します：

1. **AWS S3をDatabricks Unity Catalogに接続する方法**
2. **接続したデータをDatabricksとdbtで処理する方法**
3. **Databricksジョブを使ってデータパイプラインを自動化する方法**


## Part 1: AWS S3をUnity Catalogに接続する

Unity Catalogを利用して、AWS S3をDatabricksと連携します。この手順では、IAMロールの設定やS3バケットの準備を行います。

### 必要な準備

- **DatabricksアカウントIDの確認**:
  Databricksアカウントコンソール(アカウントを管理)から取得
  ![スクリーンショット 2024-12-04 14.46.15.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3364428/e8db9e47-f933-6d0a-b7bc-bf5bf2deac93.png)


- **AWS IAMロールの作成と設定**:
  DatabricksがS3にアクセスできるようにIAMロールを作成し、IAMポリシーとの信頼ポリシーを設定します。

- **IAMポリシーを設定**:
  必要なS3アクセス権をIAMロールに付与します。
  ![スクリーンショット 2024-12-04 14.52.07.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3364428/5e3ef9c0-f5dc-0f2e-85c8-3a9980949ffa.png)

```json
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Action": [
        "s3:GetObject",
        "s3:PutObject",
        "s3:DeleteObject",
        "s3:ListBucket",
        "s3:GetBucketLocation",
        "s3:GetLifecycleConfiguration",
        "s3:PutLifecycleConfiguration"
      ],
      "Resource": [
        "arn:aws:s3:::<作成したS3バケット>/*",
        "arn:aws:s3:::<作成したS3バケット>"
      ],
      "Effect": "Allow"
    },
    {
      "Action": [
        "sts:AssumeRole"
      ],
      "Resource": [
        "arn:aws:iam::<AWSアカウントID>:role/<作成したIAMロール>"
      ],
      "Effect": "Allow"
    }
  ]
}

```

#### 信頼ポリシーの編集

1. 作成したIAMロールを検索してクリックします。
2. 「**信頼関係**」タブを開き、「**信頼ポリシーを編集**」をクリックします。
3. 以下のポリシーを貼り付けます。
![スクリーンショット 2024-12-04 14.54.37.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3364428/bc63c646-0b23-84b4-37df-e2cc99ba6b6b.png)

```json
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "AWS": [
          "arn:aws:iam::414351767826:role/unity-catalog-prod-UCMasterRole-14S5ZJVKOTYTL",
          "arn:aws:iam::<AWSアカウントID>:role/<作成したIAMロール>"
        ]
      },
      "Action": "sts:AssumeRole",
      "Condition": {
        "StringEquals": {
          "sts:ExternalId": "<DatabricksアカウントID>"
        }
      }
    }
  ]
}
```
### Databricksでの設定

1. Databricksの「外部ロケーションを追加」画面で、S3バケットとIAMロールのARNを指定します。
![スクリーンショット 2024-12-04 14.25.10.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3364428/d3c33be3-db3e-cbd1-15c0-11f1dee559c0.png)

2. 設定を保存し、Unity Catalogの外部ストレージとしてS3を登録します。
![スクリーンショット 2024-12-04 14.25.45.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3364428/be8187ca-dca8-6aa3-ac95-d5db1e048d65.png)
　　　　　　　　　　　　　　　　　　↓
![スクリーンショット 2024-12-04 14.26.14.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3364428/ec5d7d5e-b00d-4fd6-aeb7-3b3702e928a2.png)


3.ノートブックを使った外部ロケーションからのテーブル取り込み

ノートブックから以下のコードを貼り付けます。ワークスペースとスキーマを定義して、S3の外部ロケーションからCSVを読み込み、テーブルに書き込みます。

![スクリーンショット 2025-03-18 17.21.00.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3364428/7bab4151-0a46-47fd-8dc0-3d88bf0f140f.png)


```
df = spark.read.csv("/Volumes/workspace_aws/default/test_s3_csv/customers/raw_customers.csv", header=True)
df.write.mode("overwrite").saveAsTable("default.raw_customers")

df = spark.read.csv("/Volumes/workspace_aws/default/test_s3_csv/orders/raw_orders.csv", header=True)
df.write.mode("overwrite").saveAsTable("default.raw_orders")

df = spark.read.csv("/Volumes/workspace_aws/default/test_s3_csv/payments/raw_payments.csv", header=True)
df.write.mode("overwrite").saveAsTable("default.raw_payments")
```

## Part 2: データパイプラインの構築（Databricks + dbt）

次に、S3からのデータを活用し、Databricksとdbtを使ってデータトランスフォーメーションを行います。
dbtモデルの開発をスキップしたい場合は以下のリポジトリを利用することでdbt on databricksを簡単にテストすることもできます。

デモ用のGithubリポジトリ：https://github.com/olyalukashina/jaffle_shop.git
使用の注意点として、dbt seedから初期ファイルが読み取られるようになっているため、profiles.ymlの追加と環境変数の追加を行う必要性があります。

・profiles.yml
```
jaffle_shop:
  target: dev
  outputs:
    dev:
      type: databricks
      schema: default
      host: "{{ env_var('DATABRICKS_HOST') }}"
      http_path: "{{ env_var('DATABRICKS_HTTP_PATH') }}"
      token: "{{ env_var('DATABRICKS_ACCESS_TOKEN') }}"
```
![スクリーンショット 2025-03-18 17.19.53.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3364428/46d8ac6c-af37-4ebe-b3a6-b5ee0b4a49e1.png)

・models/staging/stg_customers.sql
```
with source as (

    {#-
    Normally we would select from the table here, but we are using seeds to load
    our data in this project
    #}
    select * from default.raw_customers

),

renamed as (

    select
        id as customer_id,
        first_name,
        last_name

    from source

)

select * from renamed
```

### dbtプロジェクトの設定

1. **dbt-databricksのインストール**

   Databricksに適したパッケージをインストールします：

   ```bash
   pip install dbt-databricks
   ```

2. **dbtプロジェクトの初期化**

   既存の`jaffle_shop`プロジェクトをDatabricksに適応：

   ```bash
   dbt init my_project
   cd my_project
   ```

3. **profiles.ymlの設定**

   S3から読み込んだデータをDatabricksで処理するよう、`profiles.yml`を以下のように設定します。

   ```yaml
   my_databricks_project:
     target: dev
     outputs:
       dev:
         type: databricks
         host: <databricks-instance>
         http_path: <sql-warehouse-path>
         schema: analytics
         token: <your-access-token>
     target: dev

     default:
   ```

4. **dbtモデルの作成**

   S3データに基づくモデルを作成します（例: 売上データの集計）。

   ```sql
   -- models/sales_summary.sql
   SELECT 
       order_date,
       SUM(total) AS daily_sales
   FROM {{ ref('raw_sales') }}
   GROUP BY order_date
   ORDER BY order_date;
   ```
5. **dbtプロジェクトをGithubに保存**
  作成したdbtプロジェクトをGithubにpushします。

## Part 3: Databricksジョブで自動化

1. **Databricksジョブの作成**

   - Databricksの「ワークフロー」セクションで新しいジョブを作成して保存します。
   ![スクリーンショット 2024-12-04 14.33.29.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3364428/813c18ff-934e-f397-b0c2-85f82ab6e364.png)

   - dbtタスクを追加し、必要なdbtコマンドを設定：

   ```bash
   dbt deps
   dbt run
   ```

2. **トリガーの設定**

   - スケジュールとトリガーのタブで、S3の到着をトリガーに設定します。
   ![スクリーンショット 2024-12-04 14.21.14.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3364428/116a19be-73f1-fa21-ac03-72dbf36b7148.png)
3. **ノートブックをフローに追加**
   - これによりS3の外部ロケーションからdatabricksのテーブルへのロードも自動化されます。またこの処理が終わった後にdbtが走るようになります。
   ![スクリーンショット 2025-03-18 17.04.05.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3364428/b82b38c1-663e-4884-9705-96ad5a1d8ae5.png)

4. **動作チェック**
実際にS3にデータを入れてみてフローが動くか確認します。成功と表示され、すべてのdbtモデルがPASSされていれば成功となります。
![スクリーンショット 2024-12-04 14.37.08.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3364428/aa4e0912-c030-de68-df8c-5d7a75b248b6.png)
![スクリーンショット 2024-12-04 14.37.44.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3364428/5574f9ae-cf7b-21d4-4084-a45008122a2a.png)
またUnity Data Catalogか作られたテーブルのリネージを確認することもできます。
https://docs.databricks.com/ja/data-governance/unity-catalog/data-lineage.html
![スクリーンショット 2024-12-09 12.14.27.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3364428/68ad4f50-a9c1-3da3-bfed-fddbaec583d4.png)

3. **実行通知**
さらに以下の設定を行うことでメールやSlackへの成功失敗通知を流すことも可能です。(例：Slackの設定)
![スクリーンショット 2024-12-04 15.18.20.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3364428/152024d8-4153-24f3-794c-2f6f8149b231.png)
↓
![スクリーンショット 2024-12-04 15.17.26.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3364428/97f2dd5a-dac5-3ed7-917c-37e055511ee9.png)
成功通知
![スクリーンショット 2025-03-18 17.22.11.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3364428/73b6a2ed-dfdf-40d9-a39b-d60665966f22.png)

## 結論

この記事では、AWS S3をDatabricksと統合し、dbtを使ったデータパイプラインの構築方法を解説しました。このアプローチを活用すれば、スケーラブルなデータ基盤を構築し、データガバナンスを強化できます。さらに、Databricksジョブを使用することで、定期実行やエラー通知の自動化が可能となり、運用効率が向上します。
Databricksとdbtの連携を活用して、エンタープライズ規模のデータ分析基盤を構築してみてください！