AWSでdbt-core運用環境を導入する

## はじめに

この記事は[dbt Advent Calendar 2024](https://qiita.com/advent-calendar/2024/dbt)14日目の記事になります。

弊社では2024年5月に、DWH環境に[dbt-core](https://github.com/dbt-labs/dbt-core)を導入して、AWSの環境で運用しています。
同じようなシチュエーションでdbt-core環境を構築したい方向けに私が行った方法を記載しようと思います。参考になれば幸いです。
9割方[CloudFormation テンプレート形式](https://docs.aws.amazon.com/ja_jp/AWSCloudFormation/latest/UserGuide/template-formats.html)でコードを保存していますので、少しの手作業ですぐ環境構築ができます。

構築用のリポジトリは以下になります。forkして進めていってください。

https://github.com/nijigen-plot/dbt-athena-community-aws-resources

## 全体構成


![system_architecture.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/e111798d-ace4-3cac-8b77-11ff80f13a77.png)




- dbt-coreを用いてテーブルを更新する先は**Athena**です。[dbt-athena](https://github.com/dbt-labs/dbt-athena)とdbt-coreを一緒に使います
- **1日1回のバッチ処理**の更新を想定しています
- ソースコードの管理はGithubで行います
- 実行結果がSlackに通知されます
- ローカルからもdbtが実行可能なRoleを用意しています（これによる操作範囲はお好みで）

### ライブラリ詳細

パッケージ管理ライブラリ、[rye](https://rye.astral.sh/)を使っています。
Pythonバージョンは3.11.4で行いました。

```
dbt-athena-community==1.8.4
dbt-core==1.8.7 (dbt-athena-community 1.8.4を入れたら自動で入ります）
dbt-osmosis==0.13.2
isort==5.13.2
black==24.10.0
pre-commit==4.0.1
jinja2==3.1.4
```

ryeとリポジトリの環境は以下でインストールします。

```bash
$ curl -sSf https://rye.astral.sh/get | RYE_NO_AUTO_INSTALL=1 RYE_INSTALL_OPTION="--yes" bash
$ . "$HOME/.rye/env"
$ rye --version
$ rye sync
```

## 環境構築

ではさっそく初めて行きます。

### dbt設定

今回はAWSサンプルデータベースの[Tickit](https://docs.aws.amazon.com/ja_jp/redshift/latest/dg/c_sampledb.html)を使い、dbtを使ったデータパイプライン環境を構築します。

dbtの細かい内容についてはdbt Advent Calendarの他の方の記事で沢山上がると思いますので、こちらではサクッと進めていきます。

https://docs.aws.amazon.com/ja_jp/redshift/latest/dg/c_sampledb.html

![tickitdb.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/bba7265f-9e0c-7602-a9a6-b42067596c72.png)

7つのテーブルから成るデータ群のサンプルです。データは[ここからダウンロード可能](https://docs.aws.amazon.com/ja_jp/redshift/latest/gsg/cm-dev-t-load-sample-data.html)です。

dbtの[seed](https://docs.getdbt.com/docs/build/seeds)機能を使ってダウンロードしたデータのテーブルを作っていきます。
seedsフォルダの下に.csvファイルを置き、`dbt seed`するとテーブルを作ってくれます。

ダウンロードしたデータは最初.txtなので、csvに書き換えておきます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/5ea3753b-76fd-5e05-d905-5a4ea5f58148.png)

このように配置します。

```
└── tickit
    ├── seeds
    │   ├── allevents_pipe.csv
    │   ├── allusers_pipe.csv
    │   ├── category_pipe.csv
    │   ├── date2008_pipe.csv
    │   ├── listings_pipe.csv
    │   ├── properties.yml
    │   ├── sales_tab.csv
    │   └── venue_pipe.csv
```

それぞれのデータには、１行目に**カラムがありません**。以下をコピーして１行目に張り付けてください。

```allevents_pipe.csv
eventid|venueid|catid|dateid|eventname|starttime
```

```allusers_pipe.csv
userid|username|firstname|lastname|city|state|email|phone|likesports|liketheatre|likeconcerts|likejazz|likeclassical|likeopera|likerock|likevegas|likebroadway|likemusicals
```

```category_pipe.csv
catid|catgroup|catname|catdesc
```

```date2008_pipe.csv
dateid|caldate|day|week|month|qtr|year|holiday
```

```listings_pipe.csv
listid|sellerid|eventid|dateid|numtickets|priceperticket|totalprice|listtime
```

```sales_tab.csv
salesid	listid	sellerid	buyerid	eventid	dateid	qtysold	pricepaid	commission	saletime
```

```venue_pipe.csv
venueid|venuename|venuecity|venuestate|venueseats
```


ファイルそれぞれのdelimiterやquote、カラムの型について設定する必要がありますが、設定済みです。

```yaml:properties.yml
version: 2

seeds:
  - name: allevents_pipe
    config:
      delimiter: "|"
  - name: allusers_pipe
    config:
      delimiter: "|"
  - name: category_pipe
    config:
      delimiter: "|"
  - name: date2008_pipe
    config:
      delimiter: "|"
  - name: listings_pipe
    config:
      delimiter: "|"
      column_types:
        totalprice: decimal(8,2)
        priceperticket: decimal(8,2)
  - name: sales_tab
    config:
      delimiter: "\t"
  - name: venue_pipe
    config:
      delimiter: "|"

```

```yaml:dbt_project.yml
# 一部抜粋
seeds:
  +quote_columns: false

```

`dbt seed`は実行せず、次に進みます。




[profiles.yml](https://github.com/dbt-labs/dbt-athena?tab=readme-ov-file#configuring-your-profile)は以下のようになっています。特に弄る必要はありません。

```profiles.yml
tickit:
  outputs:
    prod:
      database: awsdatacatalog # AthenaのCatalog
      region_name: ap-northeast-1 
      s3_data_dir: "s3://{{ env_var('DBT_TABLE_RESOURCE_S3_BUCKET') }}"
      s3_staging_dir: "s3://{{ env_var('QUERY_OUTPUT_S3_BUCKET') }}"
      schema: tickit # AthenaのDatabase
      threads: 3 # 並列処理数。大きすぎるとAthenaのリソースが枯渇するので注意
      type: athena # Athenaを使う場合は"athena"
      work_group: dbt # Athenaのworkgroup
      num_retries: 0 # 実行失敗時のリトライ回数
      num_boto3_retries: 1 # boto3絡みの失敗によるリトライ回数
      aws_profile_name: "{{ env_var('AWS_PROFILE_NAME') }}" # dbt実行のaws cli profile
  target: prod
```

`profiles.yml`は、[デフォルトではホームディレクトリに置かれる](https://docs.getdbt.com/docs/core/connect-data-platform/connection-profiles)ので、リポジトリの`.dbt/profiles.yml`に置き、direnvで`DBT_PROFILES_DIR`の環境変数を書き換えて、`.dbt/profiles.yml`の設定を読むようにしています。[direnvの設定は後程説明](#.envrc設定)します。



`dbt_project.yml`の設定は以下です。
こちらも特に弄らずで大丈夫です。
```yaml:dbt_project.yml
name: 'tickit'
version: '1.0.0'
profile: 'tickit'

model-paths: ["tickit/models"]
analysis-paths: ["tickit/analyses"]
test-paths: ["tickit/tests"]
seed-paths: ["tickit/seeds"]
macro-paths: ["tickit/macros"]
snapshot-paths: ["tickit/snapshots"]

clean-targets:         # directories to be removed by `dbt clean`
  - "target"
  - "dbt_packages"

in the individual model
# files using the `{{ config(...) }}` macro.
models:
  tickit:
    +dbt-osmosis: "_{model}.yml"
    # Config indicated by + and applies to all files under models/example/
    staging:
      +materialized: view
      +schema: staging
    marts:
      +materialized: table
      +schema: marts
      +ha: true
      +format: parquet
      +write_compression: snappy
      +table_type: hive
seeds:
  +quote_columns: false
```

tickitフォルダを作って、その下に諸々入れています。
また、各model毎に[構成ファイル](https://docs.getdbt.com/reference/resource-properties/columns)が作られるように設定しています。
stagingフォルダ下はview,marts下はtableで実行されるようにしています。
構成はdbt [Best practice guides](https://docs.getdbt.com/best-practices)を参考にしました。
```
└── tickit
    ├── analyses
    ├── macros
    ├── models
    │   ├── marts
    │   │   ├── _active_user.yml
    │   │   ├── _sales.yml
    │   │   ├── active_user.sql
    │   │   └── sales.sql
    │   └── staging
    │       └── tickit
    │           ├── _stg_tickit__category.yml
    │           ├── _stg_tickit__date.yml
    │           ├── _stg_tickit__event.yml
    │           ├── _stg_tickit__listing.yml
    │           ├── _stg_tickit__sales.yml
    │           ├── _stg_tickit__users.yml
    │           ├── _stg_tickit__venue.yml
    │           ├── stg_tickit__category.sql
    │           ├── stg_tickit__date.sql
    │           ├── stg_tickit__event.sql
    │           ├── stg_tickit__listing.sql
    │           ├── stg_tickit__sales.sql
    │           ├── stg_tickit__users.sql
    │           └── stg_tickit__venue.sql
    ├── seeds
    │   ├── allevents_pipe.csv
    │   ├── allusers_pipe.csv
    │   ├── category_pipe.csv
    │   ├── date2008_pipe.csv
    │   ├── listings_pipe.csv
    │   ├── properties.yml
    │   ├── sales_tab.csv
    │   └── venue_pipe.csv
    ├── snapshots
    └── tests
```


一部紹介すると、

```sql:stg_tickit__users.sql
-- 
SELECT
    userid as user_id
    , username as user_name
    , firstname as first_name
    , lastname as last_name
    , (firstname || lastname) as full_name
    , city
    , state
    , email
    , phone as phone_number
    , likesports as like_sports
    , liketheatre as like_theatre
    , likeconcerts as like_concerts
    , likejazz as like_jazz
    , likeclassical as like_classical
    , likeopera as like_opera
    , likerock as like_rock
    , likevegas as like_vegas
    , likebroadway as like_broadway
    , likemusicals as like_musicals
FROM
    {{ source('tickit','users')}}
```

```sql:active_user.sql
SELECT
    d.calendar_date
    , COUNT(distinct u.user_id) as user_count
FROM
    {{ ref('stg_tickit__sales') }} as s
INNER JOIN
    {{ ref('stg_tickit__users')}} as u
ON
    s.buyer_id = u.user_id
INNER JOIN
    {{ ref('stg_tickit__date') }} as d
ON
    s.date_id = d.date_id
GROUP BY
    1
```

こんなかんじです。

パイプラインの構築が完了すると、Athenaでは以下のようになります。（**まだ実行しません**）

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/875e6e91-722d-29a7-8408-6a51bd49116f.png)


### AWSコンソール設定

次にAWSコンソールの設定です。これらはCloudFormationで作らないので、事前に手作業が必要です。

- ECSクラスターの作成
- S3バケットの作成
- Slack通知用Webhook URLを入れるSecret Manager
- CodePipelineとGithubの連携設定
- CloudFormationとパイプライン実行用 IAM ポリシー作成

### ECSクラスターの作成

ECSサービスの画面へ行きクラスターの作成を行います。AWS Fargateのものが既にある場合は作らなくて大丈夫です。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/52c45c1a-5470-c132-2057-c108775661ae.png)

適当な名前を入れて作成します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/459a358e-17fa-51b8-4a2d-49548988e775.png)

完成したらOKです。たまに初回は失敗するみたいです。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/b89d7aae-6e24-384e-41c0-23feb529be50.png)



#### S3バケットの作成

運用環境構築のために**バージョニングが有効になっている**S3バケットが必要ですので、1つ作成します。
**バケット名を控えておいて**ください。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/e90d050d-0705-0208-3b65-69e127d95ee1.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/f159d0f3-6178-8a86-0bed-81a4cc8dcc88.png)


#### Slack通知の準備

Slackチャンネルの右上参加人数アイコンをクリックすると上記画像がでてきます。インテグレーションから「アプリを追加する」を選択します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/3830a53a-9b86-459a-67f1-b712d117c5c1.png)

Incoming Webhooksを選択します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/41be4e6c-474d-cf1d-b851-8a11520917fd.png)



:::note warn
Incoming Webhookは現在非推奨であり、なくなる可能性があります。
> Please note, this is a legacy custom integration - an outdated way for teams to integrate with Slack. These integrations lack newer features and they will be deprecated and possibly removed in the future. We do not recommend their use. Instead, we suggest that you check out their replacement: Slack apps.
:::

画面がブラウザに飛ぶので、「追加」ボタンを選択し、チャンネルを選択、「Incoming Webhook インテグレーションの追加」を押します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/71dcbfe1-0827-4593-58d6-7666042a36e3.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/59c4d9e4-e716-1f74-781b-5517c46e4c87.png)

Webhook URLが表示されますので、これをSecrets Managerに登録します。
この時ですが、**シークレットキーはチャンネル名にしてください**。
また、secretの名前を控えておきます。(画像の場合は`slack-webhook-url`)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/9b5328ca-cbf4-541b-d220-8688a2ac419b.png)

#### CodePipelineとGithubとの接続設定

Githubの特定ブランチがPushされた時、CodePipelineが起動するようにします。

https://docs.aws.amazon.com/ja_jp/codepipeline/latest/userguide/connections-github.html

コンソールCodePipeline画面の設定→接続を設定から作ります。
基本指示に従っていけば接続の設定は特に苦戦せず作れるはずです。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/c879541c-ff07-cfb3-5765-8cf1139fb3fc.png)

次にパイプラインを作ります。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/487e2903-1366-74bb-d3d5-f0cdc145eda6.png)

Build custom pipelineを選択します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/9ba55534-2f3a-5d56-dcaa-8c81ebe1d588.png)

パイプラインの設定でパイプライン名を入れて、実行モードキューで進みます。
サービスロールは「新しいサービスロール」でロール名なしにしてサービスロール名の入力にチェックを入れれば、
`AWSCodePipelineServiceRole-{$Region}-{$PipelineName}`でロールが作られます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/4cadd631-2c62-836c-b31c-0ea0b02a4175.png)

ソースステージの設定です。

1. 「ソースプロバイダー」でGithub（バージョン2）
1. 「接続」で作成した接続を選択
1. 「リポジトリ名」はdbt-athena-community-aws-resources(一覧で出てきます)
1. 「デフォルトブランチ」はmain
1. 「トリガー」はフィルタを指定し、「プッシュ時」を選択し、「ブランチ」にmainを書く

これでdbt-athena-community-aws-resourcesのmainブランチがプッシュされた時、このCodePipelineが起動するようになります。

![202412011.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/83a37a3f-7939-5c5d-0fcd-6da5c9920f8d.png)

ビルドステージを作ります。

Other build providersを選択して、「AWS CodeBuild」を選択します。
プロジェクト名の「プロジェクトを作成する」から新しいプロジェクトを作成します。別窓が開きます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/e1a1e0d0-500e-26bc-9fa9-292b8f7ea02b.png)

オペレーティングシステムをUbuntuにします。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/e2f3173e-5f31-7c1e-4025-1ec5233ac663.png)

追加設定の環境変数で**AWS_ACCOUNT_ID : 自分のAWSアカウントIDを入れておきます**

こちらもサービスロールは自動で作られます。この後、この作られたロールに[これらポリシー](#IAMポリシー設定
)をアタッチします。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/39210fee-5dc3-2f4a-7b31-dcc3ab3a6369.png)

Buildspecは「buildspecファイルを使用する」でbuildspec.ymlと入れてください
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/06e1bb9e-c32c-b300-f42c-7084157a1e2f.png)


そのままCodePipelineに進むを押して作成を完了してください。「ビルドステージを追加する」に戻りますので、そのまま「次に」を押します。

デプロイステージを追加するは「導入段階をスキップ」で進みます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/b3626d85-aa3a-d490-e10b-83a87f01574e.png)

最後レビューで「パイプラインを作成する」で作成完了です！


#### CodeBuild_IAMポリシー設定

Codebuildで用いるロールのポリシーについては、リソースで作ることができないので事前に作成しておきます。

CodeBuildのサービスロールのポリシーは以下のようになっています。
作るのは画像の下2つです。
作ったら[先ほど作ったCodeBuild](#CodePipelineとGithubとの接続設定)のサービスロールにつけてください。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/07b85c02-7a31-59d9-2f1b-9e54b944fa63.png)


##### AmazonEC2ContainerRegistoryFullAccess

AWS Batchで使うコンテナイメージをECRにPush、リポジトリを作成するために必要です。

##### AWSCloudFormationFullAccess

CloudFormationをCodebuild上でdeployするのに必要です。

##### CodeBuildBasePolicy-dbt-athena~~~~(略)

これはCodeBuildプロジェクト作成時に自動で作られます

##### dbt-athena-community-resource-cfn-passrole

CodeBuildからCloudFormationを実行する際に必要です。
role/に続く名前は控えておいてください。

```json
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": "iam:PassRole",
            "Resource": "arn:aws:iam::xxxxxxxxxxxx:role/cloudformation-dbt-athena-community-resources-service-role"
        }
    ]
}
```

##### s3-nijipro-dbt-resources-full-access

CloudFormation Deployの為に必要なリソースをS3に置きますので、それのアクセスポリシーです。
resourceのバケット名は[最初に作ったバージョニング有のバケット](#S3バケットの作成)にしてください

```json
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "Statement1",
            "Effect": "Allow",
            "Action": [
                "s3:*"
            ],
            "Resource": [
                "arn:aws:s3:::nijipro-dbt-resources",
                "arn:aws:s3:::nijipro-dbt-resources/*"
            ]
        }
    ]
}
```

#### CloudFormation_IAMロール&ポリシー設定

[dbt-athena-community-resource-cfn-passrole](#dbt-athena-community-resource-cfn-passrole)の時に記入した名前でロールを作ります。


![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/49e4fe77-064f-e578-de0b-6283f9fa4f48.png)

ロールの信頼ポリシーは`cloudformation.amazonaws.com`に`sts:AssumeRole`を許可しています。

```json
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "Statement1",
            "Effect": "Allow",
            "Principal": {
                "Service": "cloudformation.amazonaws.com"
            },
            "Action": "sts:AssumeRole"
        }
    ]
}
```

##### cfn-nijipro-dbt-resources-create

CloudFormationによるリソース作成やロールバックに必要なアクションが含まれています。
リソースが作成された後のそれらリソースにもこのポリシーを付けていますので、実行系のアクションも含まれています。
ロールの名前は控えておいてください

ポリシー内容です。

```json
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "ecr:GetRepositoryPolicy",
                "ecr:ListImages",
                "ecr:BatchGetImage",
                "ecr:GetLifecyclePolicy",
                "ecr:GetLifecyclePolicyPreview",
                "ecr:ListTagsForResource",
                "ecr:DescribeImageScanFindings",
                "ecr:GetAuthorizationToken",
                "ecr:BatchCheckLayerAvailability",
                "ecr:GetDownloadUrlForLayer",
                "ecr:DescribeRepositories",
                "ecr:DescribeImages",
                "ecs:DeleteCluster",
                "ecs:CreateCluster",
                "ecs:DescribeClusters",
                "ecs:ListClusters",
                "ecs:TagResource",
                "ecs:UntagResource",
                "ecs:UpdateCluster",
                "ecs:DescribeContainerInstances",
                "ecs:DescribeTaskDefinition",
                "ecs:DescribeTasks",
                "ecs:ListAccountSettings",
                "ecs:ListContainerInstances",
                "ecs:ListTaskDefinitionFamilies",
                "ecs:ListTaskDefinitions",
                "ecs:ListTasks",
                "ecs:RegisterTaskDefinition",
                "ecs:DeregisterTaskDefinition",
                "ecs:RunTask",
                "ecs:StartTask",
                "ecs:StopTask",
                "ecs:UpdateContainerAgent",
                "ecs:DeregisterContainerInstance",
                "iam:CreateRole",
                "iam:PutRolePolicy",
                "iam:AttachRolePolicy",
                "iam:UpdateAssumeRolePolicy",
                "iam:CreatePolicy",
                "iam:CreatePolicyVersion",
                "iam:DetachRolePolicy",
                "iam:DeleteRole",
                "iam:DeletePolicyVersion",
                "iam:GetRole",
                "iam:GetPolicy",
                "iam:GetPolicyVersion",
                "iam:PassRole",
                "iam:ListPolicyVersions",
                "iam:DeletePolicy",
                "iam:SetDefaultPolicyVersion",
                "iam:GetInstanceProfile",
                "lambda:GetFunction",
                "lambda:GetLayerVersion",
                "lambda:GetPolicy",
                "lambda:ListFunctions",
                "lambda:ListLayers",
                "lambda:CreateFunction",
                "lambda:InvokeFunction",
                "lambda:CreateFunctionUrlConfig",
                "lambda:DeleteFunction",
                "lambda:DeleteLayerVersion",
                "lambda:DeleteFunctionUrlConfig",
                "lambda:UpdateFunctionCode",
                "lambda:UpdateFunctionConfiguration",
                "lambda:PublishVersion",
                "lambda:AddPermission",
                "lambda:RemovePermission",
                "athena:CreateWorkGroup",
                "athena:GetWorkGroup",
                "athena:DeleteWorkGroup",
                "ec2:CreateVpc",
                "ec2:ModifyVpcAttribute",
                "ec2:DeleteVpc",
                "ec2:CreateSubnet",
                "ec2:DeleteSubnet",
                "ec2:CreateInternetGateway",
                "ec2:AttachInternetGateway",
                "ec2:DetachInternetGateway",
                "ec2:DeleteInternetGateway",
                "ec2:DescribeVpcs",
                "ec2:DescribeInternetGateways",
                "ec2:DescribeSecurityGroups",
                "ec2:DescribeRouteTables",
                "ec2:DescribeNetworkAcls",
                "ec2:DescribeSubnets",
                "ec2:CreateSecurityGroup",
                "ec2:DeleteSecurityGroup",
                "ec2:AuthorizeSecurityGroupIngress",
                "ec2:AuthorizeSecurityGroupEgress",
                "ec2:RevokeSecurityGroupIngress",
                "ec2:RevokeSecurityGroupEgress",
                "ec2:CreateRouteTable",
                "ec2:DeleteRouteTable",
                "ec2:AssociateRouteTable",
                "ec2:DisassociateRouteTable",
                "ec2:CreateRoute",
                "ec2:DeleteRoute",
                "ec2:ReplaceRoute",
                "ec2:ReplaceRouteTableAssociation",
                "ec2:CreateNatGateway",
                "ec2:DeleteNatGateway",
                "ec2:DescribeNatGateways",
                "ec2:CreateNetworkAcl",
                "ec2:DeleteNetworkAcl",
                "ec2:CreateNetworkAclEntry",
                "ec2:DeleteNetworkAclEntry",
                "ec2:ReplaceNetworkAclEntry",
                "ec2:ReplaceNetworkAclAssociation",
                "ec2:CreateVpcPeeringConnection",
                "ec2:AcceptVpcPeeringConnection",
                "ec2:DeleteVpcPeeringConnection",
                "ec2:DescribeVpcPeeringConnections",
                "ec2:CreateVpnConnection",
                "ec2:DeleteVpnConnection",
                "ec2:DescribeVpnConnections",
                "ec2:CreateCustomerGateway",
                "ec2:DeleteCustomerGateway",
                "ec2:DescribeCustomerGateways",
                "ec2:CreateDhcpOptions",
                "ec2:DeleteDhcpOptions",
                "ec2:AssociateDhcpOptions",
                "ec2:DescribeDhcpOptions",
                "ec2:CreateNetworkInterface",
                "ec2:DeleteNetworkInterface",
                "ec2:DescribeNetworkInterfaces",
                "ec2:DescribeAvailabilityZones",
                "ec2:CreateTags",
                "batch:CreateComputeEnvironment",
                "batch:UpdateComputeEnvironment",
                "batch:DeleteComputeEnvironment",
                "batch:RegisterJobDefinition",
                "batch:DeregisterJobDefinition",
                "batch:SubmitJob",
                "batch:CreateJobQueue",
                "batch:UpdateJobQueue",
                "batch:DeleteJobQueue",
                "batch:TagResource",
                "batch:UntagResource",
                "batch:DescribeJobs",
                "batch:DescribeJobQueues",
                "batch:DescribeJobDefinitions",
                "batch:DescribeComputeEnvironments",
                "states:CreateStateMachine",
                "states:UpdateStateMachine",
                "states:DeleteStateMachine",
                "states:TagResource",
                "states:DescribeExecution",
                "states:DescribeStateMachine",
                "states:DescribeStateMachineForExecution",
                "states:StartExecution",
                "states:GetExecutionHistory",
                "states:StartSyncExecution",
                "states:StopExecution",
                "scheduler:CreateSchedule",
                "scheduler:UpdateSchedule",
                "scheduler:DeleteSchedule",
                "scheduler:GetSchedule",
                "scheduler:ListSchedules",
                "scheduler:TagResource",
                "scheduler:UntagResource",
                "events:PutRule",
                "events:DeleteRule",
                "events:PutTargets",
                "events:RemoveTargets",
                "events:DescribeRule",
                "events:EnableRule",
                "events:DisableRule",
                "sns:CreateTopic",
                "sns:DeleteTopic",
                "sns:SetTopicAttributes",
                "sns:GetTopicAttributes",
                "sns:Subscribe",
                "sns:AddPermission",
                "sns:RemovePermission",
                "sns:Unsubscribe",
                "s3:ListBucket",
                "s3:Put*",
                "s3:Get*",
                "s3:Delete*",
                "logs:DescribeLogGroups",
                "logs:PutLogEvents",
                "logs:CreateLogStream",
                "logs:CreateLogGroup"
            ],
            "Resource": "*"
        }
    ]
}
```

役割はそれぞれこんなかんじです。

| アクション | 用途|
| --- | --- |
| ecr | Batchによるコンテナイメージ取得 |
| ecs | Fargateを用いたBatchのタスク実行関連 |
| iam | ロールやポリシー作成削除 |
| lambda | 関数作成や実行 |
| athena | ワークグループの作成削除 |
| ec2 | BatchのVPC環境設定 |
| batch | Batch queue,definition,computing env作成やSubmit |
| states | StepFunctions ステートマシン作成削除、実行 |
| scheduler | schedulerの作成削除、実行 |
| events | Ruleの作成、イベントの送信 |
| sns | トピックの作成削除、Permissionの設定 |
| s3 | リソースの取得や書き込み |
| logs | loggroupの作成、ログの書き込み |

これでコンソール上の設定は完了です！

### リポジトリ設定

リポジトリで必要な設定です。

#### .envrc設定

各リソースは環境変数から必要な値を参照して、作るようにしています。
[direnv](https://github.com/direnv/direnv)を使って.envrcの環境変数をリポジトリ操作時に使えるようにします。


`.envrc`の定数へ文字列を入れれば、direnvとjinja2を通じて各リソースに適用されます。

```bash:.envrc
#!/bin/bash

REPOSITORY_DIR="$(pwd)"
export REPOSITORY_DIR

DBT_PROFILES_DIR="$(pwd)/.dbt"
export DBT_PROFILES_DIR

DBT_TABLE_RESOURCE_S3_BUCKET="nijigen-plot-dbt-tables"
export DBT_TABLE_RESOURCE_S3_BUCKET

QUERY_OUTPUT_S3_BUCKET="nijipro-query-results"
export QUERY_OUTPUT_S3_BUCKET

DATA_PIPELINE_RESOURCE_S3_BUCKET="nijipro-dbt-resources"
export DATA_PIPELINE_RESOURCE_S3_BUCKET

AWS_PROFILE_NAME="dbt-local"
export AWS_PROFILE_NAME

SLACK_WEBHOOK_URL_SECRETS_NAME="secretmanager-read-slack-webhook-url"
export SLACK_WEBHOOK_URL_SECRETS_NAME

SLACK_WEBHOOK_URL_CHANNEL_NAME="general"
export SLACK_WEBHOOK_URL_CHANNEL_NAME

CLOUDFORMATION_POLICY_NAME="cfn-nijipro-dbt-resources-create"
export CLOUDFORMATION_POLICY_NAME

CLOUDFORMATION_ROLE_NAME="cloudformation-dbt-athena-community-resources-service-role"
export CLOUDFORMATION_ROLE_NAME

LOCAL_DEV_ASSUME_ROLE_BEARER_USER_NAME="nijipro"
export LOCAL_DEV_ASSUME_ROLE_BEARER_USER_NAME

LOCAL_DEV_ROLE_NAME="LocalDevelopmentAssumeRole"
export LOCAL_DEV_ROLE_NAME

```


| 定数 | 用途|
| --- | --- |
| REPOSITORY_DIR | リポジトリの絶対パス|
| DIR_PROFILES_DIR | dbt profiles.yml の参照先パス |
| DBT_TABLE_RESOURCE_S3_BUCKET | dbtによるAthenaテーブル作成のS3 Locationバケット |
| QUERY_OUTPUT_S3_BUCKET | クエリ実行結果保存先バケット |
| DATA_PIPELINE_RESOURCE_S3_BUCKET | データパイプライン環境構築のためのリソースバケット。[S3バケットの作成でバージョニングを有効にして作ったもの](#S3バケットの作成) |
| AWS_PROFILE_NAME | ローカル開発用のAWSプロファイル名 このままでもOK |
| SLACK_WEBHOOK_URL_SECRETS_NAME | [作成したSlack Webhook URLのSecrets Manager名](#Slack通知の準備) |
| SLACK_WEBHOOK_URL_CHANNEL_NAME | [作成したSlack Webhook URLのSecrets Managerのチャンネル名](#Slack通知の準備) |
| CLOUDFORMATION_POLICY_NAME | [先ほど作成した](#CloudFormation_IAMロール&ポリシー設定)CloudFormation用ロールにアタッチするポリシー名 |
| CLOUDFORMATION_ROLE_NAME | [先ほど作成した](#CloudFormation_IAMロール&ポリシー設定)CloudFormationにアタッチするロール名 |
| LOCAL_DEV_ASSUME_ROLE_BEARER_USER_NAME | ローカル開発用ロールをAssume RoleするAWS User名 |
| LOCAL_DEV_ROLE_NAME | Assume Roleさせるロール名 このままでもOK|


最後にdirenvをインストールして、適用します。

```bash
# direnvインストールと適用
$ curl -sfL https://direnv.net/install.sh | bash
$ eval "$(direnv hook bash)"
$ direnv allow
```

設定が完了したら、`rye run python ./script/jinja_render.py`を実行しコミットして、mainにプッシュしましょう！


## 動作状況チェック

CodePipelineが起動し、CloudFormationからリソースが作成されます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/a72d3ca1-861a-e1b4-cfc0-a48bee10393a.png)


ローカル開発用ロールをAssume RoleするAWS User名でaws cliが実行できるようにしておき、`$ bash ./script/dbt-local-assume-role.sh`を実行すると、ローカル開発用のAWSプロファイル名でセッションが作られ、ローカル環境からdbt seed,dbt run等のコマンドが実行できるようになります！

さっそくdbt seedしてみます。
```bash
$ bash ./script/dbt-local-assume-role.sh 
quarkgabber@DESKTOP-AAUG4GU ~/Repositories/dbt-athena-community-aws-resources (main $) 
$ rye run dbt seed
16:06:29  Running with dbt=1.8.7
16:06:29  Registered adapter: athena=1.8.4
16:06:30  Found 9 models, 7 seeds, 3 data tests, 455 macros
16:06:30  
16:06:32  Concurrency: 3 threads (target='prod')
16:06:32  
16:06:32  1 of 7 START seed file tickit.allevents_pipe ................................... [RUN]
16:06:32  2 of 7 START seed file tickit.allusers_pipe .................................... [RUN]
16:06:32  3 of 7 START seed file tickit.category_pipe .................................... [RUN]
 16:06:51  2 of 7 OK loaded seed file tickit.allusers_pipe ................................ [INSERT 49990 in 18.59s]
16:06:51  4 of 7 START seed file tickit.date2008_pipe .................................... [RUN]
16:06:51  3 of 7 OK loaded seed file tickit.category_pipe ................................ [INSERT 11 in 18.63s]
16:06:51  5 of 7 START seed file tickit.listings_pipe .................................... [RUN]
16:06:52  1 of 7 OK loaded seed file tickit.allevents_pipe ............................... [INSERT 8798 in 20.10s]
16:06:52  6 of 7 START seed file tickit.sales_tab ........................................ [RUN]
16:07:32  6 of 7 OK loaded seed file tickit.sales_tab .................................... [INSERT 172462 in 40.28s]
16:07:32  7 of 7 START seed file tickit.venue_pipe ....................................... [RUN]
16:07:33  4 of 7 OK loaded seed file tickit.date2008_pipe ................................ [INSERT 365 in 42.33s]
16:07:46  5 of 7 OK loaded seed file tickit.listings_pipe ................................ [INSERT 192500 in 54.93s]
16:07:46  7 of 7 OK loaded seed file tickit.venue_pipe ................................... [INSERT 202 in 13.40s]
16:07:46  
16:07:46  Finished running 7 seeds in 0 hours 1 minutes and 16.10 seconds (76.10s).
16:07:46  
16:07:46  Completed successfully
16:07:46  
16:07:46  Done. PASS=7 WARN=0 ERROR=0 SKIP=0 TOTAL=7
```

テーブルが作成されました。

毎朝日本時間の8時に、EventBridge SchedulerがStepFunctionsを実行し、dbtコマンド群が走ります

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/22668226-eb1d-0f97-acdf-e240ae1bb635.png)



その後、結果がSNSとLambdaを介してSlackに通知されます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/4dbb647c-9c12-d907-8c32-effb689b8a92.png)


[全体構成](#全体構成)で紹介した運用環境ができました！
リポジトリのmodelやseedを、実際に使うデータに書き換えればこのまま運用できます。


### その他リポジトリ内容

これらは弄らなくても特に問題無い、[動作状況チェック](#動作状況チェック)の裏側の説明です。
興味がない方は飛ばしても大丈夫です。


#### buildspec.j2.yml



```yaml:buildspec.j2.yml
version: 0.2

env:
  variables:
    data_pipeline_resource_s3_bucket: {{ data_pipeline_resource_s3_bucket }}
    cloudformation_role_name: {{ cloudformation_role_name }}
    dwh_stack_name: 'dbt-infrastructure'

phases:
  install:
    runtime-versions:
      python: 3.11.4
    commands:
      - python3 --version
      - sudo apt-get update
      # ryeを入れる
      - curl -sSf https://rye.astral.sh/get | RYE_NO_AUTO_INSTALL=1 RYE_INSTALL_OPTION="--yes" bash
      - . "$HOME/.rye/env"
      - rye --version
      - rye sync --no-dev --no-lock
  pre_build:
    commands:
      - aws --version
      # StepFunctions yamlファイルをS3へsync
      - aws s3 sync --exact-timestamps --delete aws/stepfunctions/ s3://$data_pipeline_resource_s3_bucket/stepfunctions/
      # Lambda lambda_function.pyをZip化し、S3へsync
      - bash script/create_zip_files.sh
      # StepFunctions,Lambdaリソースの最新バージョンIDを取得
      - AWSServiceExecuteResultNotificationParam=$(aws s3api list-object-versions --bucket $data_pipeline_resource_s3_bucket --prefix lambda/aws_service_execute_result_notification.zip | jq -r '.Versions[0].VersionId')
      - DbtBatchParam=$(aws s3api list-object-versions --bucket $data_pipeline_resource_s3_bucket --prefix stepfunctions/dbt_batch.yml | jq -r '.Versions[0].VersionId')
  build:
    commands:
      - ORIGINAL_DIR=$(pwd)
      - echo "dbt Docker Image Build & Push"
      - echo $AWS_ACCOUNT_ID
      # Batch用イメージをPush
      - docker build -t dbt-batch .
      - >
        aws ecr get-login-password --region ap-northeast-1 |
        docker login --username AWS --password-stdin $AWS_ACCOUNT_ID.dkr.ecr.ap-northeast-1.amazonaws.com
      - docker tag dbt-batch:latest $AWS_ACCOUNT_ID.dkr.ecr.ap-northeast-1.amazonaws.com/dbt-batch:latest
      - docker push $AWS_ACCOUNT_ID.dkr.ecr.ap-northeast-1.amazonaws.com/dbt-batch:latest
      - cd $ORIGINAL_DIR
  post_build:
    commands:
      # CloudFormation スタックDeploy
      - echo deploy data pipeline infrastructure
      - aws cloudformation package --template-file aws/cloudformation/$dwh_stack_name.yml --s3-bucket $data_pipeline_resource_s3_bucket --s3-prefix cloudformation/package --output-template-file /tmp/$dwh_stack_name.yml --force-upload
      - |
        aws cloudformation deploy --no-fail-on-empty-changeset --template-file /tmp/$dwh_stack_name.yml --stack-name $dwh_stack_name --role-arn arn:aws:iam::$AWS_ACCOUNT_ID:role/$cloudformation_role_name --capabilities CAPABILITY_NAMED_IAM CAPABILITY_IAM \
        --parameter-overrides \
        "AWSServiceExecuteResultNotificationParam=$AWSServiceExecuteResultNotificationParam" \
        "DbtBatchParam=$DbtBatchParam" \

```

Lamdba,StepFunctionsのリソースは、S3のバージョニングを有効化して、バージョンIDを取得、それをAWS::StepFunctions::StateMachineの[DefinitionS3Location](https://docs.aws.amazon.com/ja_jp/AWSCloudFormation/latest/UserGuide/aws-properties-stepfunctions-statemachine-s3location.html)のVersionとAWS::Lambda::Functionの[S3ObjectVersion](https://docs.aws.amazon.com/ja_jp/AWSCloudFormation/latest/UserGuide/aws-properties-lambda-function-code.html#cfn-lambda-function-code-s3objectversion)に反映します。

そうしないと、コードを変更してもリソースの変更が検知されず、**修正の際コードの定義と実際の定義が違うくなるから**です。

#### dbt_batch.j2.yml


dbt_batch.j2.ymlはStepFunctionsステートマシンの定義ファイルです。

dbt run → dbt docs generate(静的コンテンツも書き出す) →dbt testを行います。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/5dad729c-8691-0797-f022-f8091e363eba.png)

```yaml:dbt_batch.j2.yml
Comment: dbt workflow with AWS Batch
StartAt: dbt run
States:
  dbt run:
    Type: Task
    Resource: arn:aws:states:::batch:submitJob.sync
    Parameters:
      JobName: dbt-run-daily
      ContainerOverrides:
        Command:
          - "/bin/bash"
          - "-c"
          - "source $HOME/.rye/env && rye run dbt run"
      JobQueue: dbt-batch-job-queue
      JobDefinition: dbt-batch-job-definition
    Next: dbt docs generate
    Catch:
      - ErrorEquals:
          - States.ALL
        Next: Fail
        ResultPath: $.Error
  dbt docs generate:
    Type: Task
    Resource: arn:aws:states:::batch:submitJob.sync
    Parameters:
      JobName: dbt-docs-generate
      ContainerOverrides:
        Command:
          - "/bin/bash"
          - "-c"
          - "source $HOME/.rye/env && rye run dbt docs generate && rye run python generate_static_html.py && aws s3 cp target/index2.html s3://{{ data_pipeline_resource_s3_bucket }}/docs.html"
      JobQueue: dbt-batch-job-queue
      JobDefinition: dbt-batch-job-definition
    Next: dbt test
    Catch:
      - ErrorEquals:
          - States.ALL
        Next: Fail
        ResultPath: $.Error
  dbt test:
    Type: Task
    Resource: arn:aws:states:::batch:submitJob.sync
    Parameters:
      JobName: dbt-test-daily
      ContainerOverrides:
        Command:
          - "/bin/bash"
          - "-c"
          - "source $HOME/.rye/env && rye run dbt test"
      JobQueue: dbt-batch-job-queue
      JobDefinition: dbt-batch-job-definition
    Next: Success
    Catch:
      - ErrorEquals:
          - States.ALL
        Next: Fail
        ResultPath: $.Error
  Success:
    Type: Succeed
  Fail:
    Type: Fail
    ErrorPath: $.Error
    CausePath: $.Error


```


> 静的コンテンツも書き出す

ですが、[S3へdocsのhtmlファイルを公開して閲覧できるよう](https://github.com/nijigen-plot/dbt-athena-community-aws-resources/blob/main/generate_static_html.py)にしています。
`s3://{{ data_pipeline_resource_s3_bucket }}/docs.html`に配置されます。

S3のバケットポリシーなどを利用して、社内用のドキュメントにできます！

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/94ae7f1b-5635-ed8f-dc21-22dbcd93aeba.png)


#### lambda function

[StepFunctionの実行結果を受け取りSlackへ通知する関数](https://github.com/nijigen-plot/dbt-athena-community-aws-resources/blob/main/aws/lambda/aws_service_execute_result_notification/lambda_function.py)です。
その他、EventBridge RuleへAthenaやGlue Jobを追加することで、それらサービスでも通知することが可能です。

実行後はこんなかんじでSlackに通知が来ます。

![202412012.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/e1bf2697-433f-4d02-2de2-bc52e89755f9.png)


#### dbt-infrastructure.j2.yml

CloudFormation テンプレートファイルです。
ファイル名をj2.ymlとしているのは[cfn-lint](https://github.com/aws-cloudformation/cfn-lint)が効かなくなるからです。

抜粋して解説しようと思います。全コードは以下です。

https://github.com/nijigen-plot/dbt-athena-community-aws-resources/blob/main/aws/cloudformation/dbt-infrastructure.j2.yml


CFnで作られた各リソースにアタッチするロールです。
CFnにアタッチするポリシーを流用しています。信頼ポリシーを必要最小限にしています。
```yaml
  DataPipeLineOperationRole:
    Type: AWS::IAM::Role
    DeletionPolicy: Delete
    UpdateReplacePolicy: Delete
    Properties:
      RoleName: DataPipeLineOperationRole
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Principal:
              Service:
                - states.amazonaws.com
                - scheduler.amazonaws.com
                - sns.amazonaws.com
                - glue.amazonaws.com
                - lambda.amazonaws.com
                - events.amazonaws.com
                - batch.amazonaws.com
                - ecs-tasks.amazonaws.com
            Action: sts:AssumeRole
      ManagedPolicyArns:
        - !Sub arn:${AWS::Partition}:iam::${AWS::AccountId}:policy/${CloudFormationPolicyName}
        - !Ref SlackWebhookURLSecrets
        - !Ref AthenaAccess
```


ローカル環境でdbtを実行するためのロールとポリシーです。

```yaml
  LocalDevelopmentAssumeRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: !Sub ${LocalDevRoleName}
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Principal:
              AWS:
                - !Sub arn:${AWS::Partition}:iam::${AWS::AccountId}:user/${LocalDevAssumeRoleBearerUserName}
            Action: sts:AssumeRole
      ManagedPolicyArns:
        - !Ref AthenaAccess
  AthenaAccess:
    Type: AWS::IAM::ManagedPolicy
    DeletionPolicy: Delete
    UpdateReplacePolicy: Delete
    Properties:
      PolicyDocument:
        Version: 2012-10-17
        Statement:
          - Action:
              - s3:*
            Resource:
              - !Sub arn:${AWS::Partition}:s3:::${DbtTableResourceS3Bucket}
              - !Sub arn:${AWS::Partition}:s3:::${DbtTableResourceS3Bucket}/*
            Effect: Allow
          - Action:
              - s3:GetBucketLocation
              - s3:GetObject
              - s3:ListBucket
              - s3:ListBucketMultipartUploads
              - s3:ListMultipartUploadParts
              - s3:AbortMultipartUpload
              - s3:PutObject
              - s3:DeleteObject
            Resource:
              - !Sub arn:${AWS::Partition}:s3:::${QueryOutputS3Bucket}
              - !Sub arn:${AWS::Partition}:s3:::${QueryOutputS3Bucket}/*
            Effect: Allow
          - Action:
              - athena:StartQueryExecution
              - athena:GetQueryResults
              - athena:DeleteNamedQuery
              - athena:GetNamedQuery
              - athena:ListQueryExecutions
              - athena:StopQueryExecution
              - athena:GetQueryResultsStream
              - athena:ListNamedQueries
              - athena:CreateNamedQuery
              - athena:GetQueryExecution
              - athena:BatchGetNamedQuery
              - athena:BatchGetQueryExecution
              - athena:UpdateWorkGroup
              - athena:GetWorkGroup
              - athena:CreatePreparedStatement
              - athena:DeletePreparedStatement
              - athena:GetPreparedStatement
              - athena:UpdatePreparedStatement
            Resource:
              - !Sub arn:${AWS::Partition}:athena:ap-northeast-1:${AWS::AccountId}:workgroup/${AthenaDbtWorkgroup}
            Effect: Allow
          - Action:
              - glue:Get*
              - glue:Create*
              - glue:Delete*
              - glue:Describe*
              - glue:List*
              - glue:SearchTables
              - glue:Update*
            Resource:
              - !Sub arn:${AWS::Partition}:glue:ap-northeast-1:${AWS::AccountId}:catalog
              - !Sub arn:${AWS::Partition}:glue:ap-northeast-1:${AWS::AccountId}:database/*
              - !Sub arn:${AWS::Partition}:glue:ap-northeast-1:${AWS::AccountId}:table/*
            Effect: Allow
          - Action:
              - athena:GetDataCatalog
              - athena:GetDatabase
              - athena:GetTableMetadata
              - athena:ListDatabases
              - athena:ListTableMetadata
              - athena:UpdateDataCatalog
            Resource: !Sub arn:${AWS::Partition}:athena:ap-northeast-1:${AWS::AccountId}:datacatalog/*
            Effect: Allow
```

#### dbt-local-assume-role.sh
ローカル実行用Assume Role有効化スクリプトです。

実行すると、[.envrc](#.envrc設定)に記載したAssumeRoleするユーザー名に対して信頼ポリシーで許可され、[profile.yml](#dbt設定)のaws_profile_nameと同じprofile名のcredentialsができます。


```bash:dbt-local-assume-role.sh
#! /bin/bash

AWS_ACCOUNT_ID=$(aws sts get-caller-identity --query "Account" --output text)
ROLE_ARN="arn:aws:iam::$AWS_ACCOUNT_ID:role/$LOCAL_DEV_ROLE_NAME"
SESSION_NAME="$AWS_PROFILE_NAME"
PROFILE_NAME="$AWS_PROFILE_NAME"

CREDENTIALS=$(aws sts assume-role --role-arn "$ROLE_ARN" --role-session-name "$SESSION_NAME" )

ACCESS_KEY=$(echo "$CREDENTIALS" | jq -r '.Credentials.AccessKeyId')
SECRET_KEY=$(echo "$CREDENTIALS" | jq -r '.Credentials.SecretAccessKey')
SESSION_TOKEN=$(echo "$CREDENTIALS" | jq -r '.Credentials.SessionToken')

# ~/.aws/credentialsファイルに認証情報を追加または更新
aws configure set aws_access_key_id "$ACCESS_KEY" --profile "$PROFILE_NAME"
aws configure set aws_secret_access_key "$SECRET_KEY" --profile "$PROFILE_NAME"
aws configure set aws_session_token "$SESSION_TOKEN" --profile "$PROFILE_NAME"
aws configure set region ap-northeast-1 --profile "$PROFILE_NAME"

```

セッション時間はデフォルト1時間ですが、変更も可能です。

https://docs.aws.amazon.com/ja_jp/AWSCloudFormation/latest/UserGuide/aws-resource-iam-role.html#cfn-iam-role-maxsessionduration

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/68f8a6f4-8e40-b080-c510-72ba597e4603.png)


#### jinja_render.py

cloudformation,codebuildのjinja テンプレートファイルからレンダリングしてj2なしファイルに書き出します。

```python:jinja_render.py
import os

from jinja2 import Environment, FileSystemLoader

env = Environment(loader=FileSystemLoader("."))

# cloudformation yaml template
cfn_template = env.get_template("./aws/cloudformation/dbt-infrastructure.j2.yml")
buildspec_template = env.get_template("./buildspec.j2.yml")
stepfunctions_template = env.get_template("./aws/stepfunctions/dbt_batch.j2.yml")

params = {
    "dbt_table_resource_s3_bucket": os.getenv("DBT_TABLE_RESOURCE_S3_BUCKET"),
    "query_output_s3_bucket": os.getenv("QUERY_OUTPUT_S3_BUCKET"),
    "data_pipeline_resource_s3_bucket": os.getenv("DATA_PIPELINE_RESOURCE_S3_BUCKET"),
    "slack_webhook_url_secrets_name": os.getenv("SLACK_WEBHOOK_URL_SECRETS_NAME"),
    "slack_webhook_url_channel_name": os.getenv("SLACK_WEBHOOK_URL_CHANNEL_NAME"),
    "cloudformation_policy_name": os.getenv("CLOUDFORMATION_POLICY_NAME"),
    "cloudformation_role_name": os.getenv("CLOUDFORMATION_ROLE_NAME"),
    "local_dev_assume_role_bearer_user_name": os.getenv(
        "LOCAL_DEV_ASSUME_ROLE_BEARER_USER_NAME"
    ),
    "local_dev_role_name": os.getenv("LOCAL_DEV_ROLE_NAME"),
}

with open(
    f'{os.getenv("REPOSITORY_DIR", os.path.expanduser("."))}/aws/cloudformation/dbt-infrastructure.yml',
    "w",
) as file:
    file.write(cfn_template.render(params) + "\n")
with open(
    f'{os.getenv("REPOSITORY_DIR", os.path.expanduser("."))}/buildspec.yml', "w"
) as file:
    file.write(buildspec_template.render(params) + "\n")
with open(
    f'{os.getenv("REPOSITORY_DIR", os.path.expanduser("."))}/aws/stepfunctions/dbt_batch.yml',
    "w",
) as file:
    file.write(stepfunctions_template.render(params) + "\n")


```


変更時は`$ rye run python script/jinja_render.py`の実行が必須ですので、実行されているかはGithub ActionsのCIでチェックしています。


#### Github Actions CI

pre-commitによるリンターフォーマッターの実行と、jinja_render.pyが実行されているかの確認を行っています。mainブランチへPR時に走るので、mainブランチ直Pushは出来ない想定です。

```yaml:lint.yml
name: lint

on:
  pull_request:
    branches:
      - main

jobs:
  lint:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4.1.1
        with:
          fetch-depth: 2
      - name: Set up Python
        uses: actions/setup-python@v5.1.0
        with:
          python-version: "3.11.4"
      - name: Install pre-commit hooks
        run: |
          pip install -U pip==24.0
          pip install -U pre-commit==4.0.1
          pre-commit install
      - name: Run pre-commit hooks
        run: |
          pre-commit run --all-files
      - name: Install libraries and run jinja_render
        run: |
          pip install -U jinja2==3.1.4
          mkdir -p ~/.local/bin
          curl -sfL https://direnv.net/install.sh | bash
          direnv allow
      - name: Load PATH changes
        run: direnv exec . sh -c 'echo $PATH' > "$GITHUB_PATH"
      - name: Load other environment changes
        run: direnv export gha >> "$GITHUB_ENV"
      - name: Run jinja_render
        run: |
          mv aws/cloudformation/dbt-infrastructure.yml /tmp/dbt-infrastructure.yml
          mv buildspec.yml /tmp/buildspec.yml
          mv aws/stepfunctions/dbt_batch.yml /tmp/dbt_batch.yml
          python script/jinja_render.py
      - name: jinja render check
        run: |
          diff aws/cloudformation/dbt-infrastructure.yml /tmp/dbt-infrastructure.yml || (echo "dbt-infrastructure.yml is out of date. Run 'rye run python ./script/jinja_render.py' to update it." && exit 1)
          diff buildspec.yml /tmp/buildspec.yml || (echo "buildspec.yml is out of date. Run 'rye run python ./script/jinja_render.py' to update it." && exit 1)
          diff aws/stepfunctions/dbt_batch.yml /tmp/dbt_batch.yml || (echo "dbt_batch.yml is out of date. Run 'rye run python ./script/jinja_render.py' to update it." && exit 1)


```

```yaml:.pre-commit-config.yaml
repos:
  - repo: https://github.com/pre-commit/pre-commit-hooks
    rev: v4.6.0
    hooks:
      - id: end-of-file-fixer
      - id: trailing-whitespace
      - id: check-json
      - id: check-yaml
        exclude: |
          (?x)^(
            ^(aws/cloudformation/.*\.yml$) |
            ^(buildspec\.j2\.yml$)
          )$
      - id: detect-private-key
      - id: debug-statements
  - repo: https://github.com/shellcheck-py/shellcheck-py
    rev: v0.10.0.1
    hooks:
      - id: shellcheck
  - repo: https://github.com/psf/black
    rev: 24.10.0
    hooks:
      - id: black
        args: ["--config", "pyproject.toml"]
  - repo: https://github.com/pycqa/isort
    rev: 5.13.2
    hooks:
      - id: isort
        args: ["-sp", "pyproject.toml"]
  # - repo: https://github.com/dbt-checkpoint/dbt-checkpoint
  #   rev: 2.0.6
  #   hooks:
  #     - id: dbt-parse
exclude: ^(.*\.drawio$)

```





## その他便利スクリプト&AWSで作っているもの

ここからは、パイプライン構築でその他作っているものや、便利スクリプトについて簡単に紹介したいと思います。

### Google Drive / Sheets ETL

![lamdba_glue_job.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/83fdb577-ac8e-e737-f43e-fb1704765a87.png)


Google Driveのファイル群・特定ファイル・Google Spreadsheetの特定シートからLambda→S3→Glue Job→AthenaのETLパイプラインを作っています。その後[dbt source](https://docs.getdbt.com/docs/build/sources)に追加して管理しています。

これらの流れも、StepFunctionsでラップすることにより、EventBridge Ruleで実行結果をキャッチできるので、Slackで通知が可能です。

Glue Jobを使う理由ですが、「運用上スプレッドシートやGoogle Driveが好ましいが、クラウド同期もしたい」場合、カラム名やセルの型チェックなどをPythonで仕込み、エラーハンドリングする必要があるからです。

### Lambda→SQS→Athena(S3) データ取得

![sqs_lambda.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/318b021a-7221-30a9-f220-3b55a4801bd2.png)


APIを沢山飛ばすときは、APIコード生成→SQSキュー生成Lambdaとキュー受け取りLambdaを作ってS3にデータを入れ、Athenaで読み取ってテーブルを作っています。SQSを使うことにより再実行もできますし、デッドレターキューもあるので便利です。これもその後dbt sourceでdbt管理下に置いています。

### AWS Cloudwatch Logs 監視用ツール

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/9f42e6aa-e46c-87ad-0c62-1862feb41439.png)


@ktatさんが作成したaws-cliラッパースクリプト[aws-logs](https://qiita.com/ktat/items/6494390dec4b5b9c662b#aws-logs)を使っています。

Batch処理中の状況を見たい時、処理時エラーの内容を見たい時など便利です。
Batchはジョブ定義でLogGroupを指定しないと`/aws/batch/job`にすべてのジョブのログが溜まります。
今回のCFnテンプレートでも/aws/batch/jobに溜まるので、別の場所にしたい場合は[LogConfiguration](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-batch-jobdefinition-logconfiguration.html)で

```
LogDriver: awslogs
Options: 
  - awslogs-group: /dbt/job
```

のようにすると良いと思います。
[AWS::Logs::LogGroup](https://docs.aws.amazon.com/ja_jp/AWSCloudFormation/latest/UserGuide/aws-resource-logs-loggroup.html)の追加も必要です。
あとは[CFnポリシー](#IAMポリシー設定)に[CloudWatch Logs](#IAMポリシー設定)の`DeleteLogGroup,DescribeLogGroups,ListTagsLogGroup`等追加が必要です（ロールバック時失敗するから）

### AWS Athena クエリ結果表示ツール

CLIからコマンドでクエリ結果を表示したり、.sqlファイルから表示したり、2つのテーブルの集計を比較することができます。これは私がつくりました。


https://qiita.com/nijigen_plot/items/bf8ec26d7226d1e7eb6d

https://github.com/nijigen-plot/aws-cli-data-pipeline-tools

いちいちコンソールに行かなくてもクエリを書いて結果を得られるので便利です。Dev環境とProd環境のデータチェックにも使えます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/e18790de-cf50-4db2-9ae6-0cf4dec83bd6.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/635079/b4f04224-21cc-f5e2-757f-14a976c255cc.png)

### 非管理テーブル検出ツール

dbtとAthena(Glue)で片方にしか存在しないテーブルをチェックします。
消し忘れや、dbt登録忘れのチェックに良いです。

https://qiita.com/nijigen_plot/items/4005452818c39408f8aa

## さいごに

ここまでみていただきありがとうございました。
自分自身でも1からAWSアカウントを作って動作確認していますので、導入を検討している型は是非試してみてください！

分からない点や、フィードバック、こうすると良くなる等アドバイスいただけますと嬉しいです。

また、[株式会社WANOアドベントカレンダー](https://qiita.com/advent-calendar/2024/wano-group)ではデータパイプライン及びデータ利活用での非技術的な話を書きましたので、もしよければそちらも見ていただけると嬉しいです。

https://qiita.com/nijigen_plot/items/c05dd196c067df4b7e1e?fbclid=IwZXh0bgNhZW0CMTEAAR3sbn2tHgFenG8RilSz6PYYe3jcwVIrJycDgVHiebd_6LOnGROuPyBZ_I0_aem_-jeTJ5vHzD8UFCNUOD973A