はじめに

この記事はdbt Advent Calendar 202414日目の記事になります。

弊社では2024年5月に、DWH環境にdbt-coreを導入して、AWSの環境で運用しています。
同じようなシチュエーションでdbt-core環境を構築したい方向けに私が行った方法を記載しようと思います。参考になれば幸いです。
9割方CloudFormation テンプレート形式でコードを保存していますので、少しの手作業ですぐ環境構築ができます。

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

全体構成

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

ライブラリ詳細

パッケージ管理ライブラリ、ryeを使っています。
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とリポジトリの環境は以下でインストールします。

$ 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を使い、dbtを使ったデータパイプライン環境を構築します。

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

7つのテーブルから成るデータ群のサンプルです。データはここからダウンロード可能です。

dbtのseed機能を使ってダウンロードしたデータのテーブルを作っていきます。
seedsフォルダの下に.csvファイルを置き、dbt seedするとテーブルを作ってくれます。

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

このように配置します。

└── 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、カラムの型について設定する必要がありますが、設定済みです。

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: "|"

dbt_project.yml

# 一部抜粋
seeds:
  +quote_columns: false

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

profiles.ymlは以下のようになっています。特に弄る必要はありません。

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は、デフォルトではホームディレクトリに置かれるので、リポジトリの.dbt/profiles.ymlに置き、direnvでDBT_PROFILES_DIRの環境変数を書き換えて、.dbt/profiles.ymlの設定を読むようにしています。direnvの設定は後程説明します。

dbt_project.ymlの設定は以下です。
こちらも特に弄らずで大丈夫です。

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毎に構成ファイルが作られるように設定しています。
stagingフォルダ下はview,marts下はtableで実行されるようにしています。
構成はdbt Best practice guidesを参考にしました。

└── 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

一部紹介すると、

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')}}

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では以下のようになります。（まだ実行しません）

AWSコンソール設定

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

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

ECSクラスターの作成

ECSサービスの画面へ行きクラスターの作成を行います。AWS Fargateのものが既にある場合は作らなくて大丈夫です。

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

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

S3バケットの作成

運用環境構築のためにバージョニングが有効になっているS3バケットが必要ですので、1つ作成します。
バケット名を控えておいてください。

Slack通知の準備

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

Incoming Webhooksを選択します。

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 インテグレーションの追加」を押します。

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

CodePipelineとGithubとの接続設定

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

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

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

Build custom pipelineを選択します。

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

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

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

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

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

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

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

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

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

Buildspecは「buildspecファイルを使用する」でbuildspec.ymlと入れてください

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

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

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

CodeBuild_IAMポリシー設定

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

CodeBuildのサービスロールのポリシーは以下のようになっています。
作るのは画像の下2つです。
作ったら先ほど作ったCodeBuildのサービスロールにつけてください。

AmazonEC2ContainerRegistoryFullAccess

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

AWSCloudFormationFullAccess

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

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

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

dbt-athena-community-resource-cfn-passrole

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

{
    "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のバケット名は最初に作ったバージョニング有のバケットにしてください

{
    "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の時に記入した名前でロールを作ります。

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

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

cfn-nijipro-dbt-resources-create

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

ポリシー内容です。

{
    "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を使って.envrcの環境変数をリポジトリ操作時に使えるようにします。

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

.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バケットの作成でバージョニングを有効にして作ったもの
AWS_PROFILE_NAME	ローカル開発用のAWSプロファイル名このままでもOK
SLACK_WEBHOOK_URL_SECRETS_NAME	作成したSlack Webhook URLのSecrets Manager名
SLACK_WEBHOOK_URL_CHANNEL_NAME	作成したSlack Webhook URLのSecrets Managerのチャンネル名
CLOUDFORMATION_POLICY_NAME	先ほど作成したCloudFormation用ロールにアタッチするポリシー名
CLOUDFORMATION_ROLE_NAME	先ほど作成したCloudFormationにアタッチするロール名
LOCAL_DEV_ASSUME_ROLE_BEARER_USER_NAME	ローカル開発用ロールをAssume RoleするAWS User名
LOCAL_DEV_ROLE_NAME	Assume Roleさせるロール名このままでもOK

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

# 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からリソースが作成されます。

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

さっそくdbt seedしてみます。

$ 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コマンド群が走ります

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

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

その他リポジトリ内容

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

buildspec.j2.yml

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のVersionとAWS::Lambda::FunctionのS3ObjectVersionに反映します。

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

dbt_batch.j2.yml

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

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

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ファイルを公開して閲覧できるようにしています。
s3://{{ data_pipeline_resource_s3_bucket }}/docs.htmlに配置されます。

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

lambda function

StepFunctionの実行結果を受け取りSlackへ通知する関数です。
その他、EventBridge RuleへAthenaやGlue Jobを追加することで、それらサービスでも通知することが可能です。

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

dbt-infrastructure.j2.yml

CloudFormation テンプレートファイルです。
ファイル名をj2.ymlとしているのはcfn-lintが効かなくなるからです。

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

CFnで作られた各リソースにアタッチするロールです。
CFnにアタッチするポリシーを流用しています。信頼ポリシーを必要最小限にしています。

  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を実行するためのロールとポリシーです。

  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に記載したAssumeRoleするユーザー名に対して信頼ポリシーで許可され、profile.ymlのaws_profile_nameと同じprofile名のcredentialsができます。

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時間ですが、変更も可能です。

jinja_render.py

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

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は出来ない想定です。

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)

.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

Google Driveのファイル群・特定ファイル・Google Spreadsheetの特定シートからLambda→S3→Glue Job→AthenaのETLパイプラインを作っています。その後dbt sourceに追加して管理しています。

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

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

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

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

AWS Cloudwatch Logs 監視用ツール

@ktatさんが作成したaws-cliラッパースクリプトaws-logsを使っています。

Batch処理中の状況を見たい時、処理時エラーの内容を見たい時など便利です。
Batchはジョブ定義でLogGroupを指定しないと/aws/batch/jobにすべてのジョブのログが溜まります。
今回のCFnテンプレートでも/aws/batch/jobに溜まるので、別の場所にしたい場合はLogConfigurationで

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

のようにすると良いと思います。
AWS::Logs::LogGroupの追加も必要です。
あとはCFnポリシーにCloudWatch LogsのDeleteLogGroup,DescribeLogGroups,ListTagsLogGroup等追加が必要です（ロールバック時失敗するから）

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

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

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

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

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

さいごに

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

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

また、株式会社WANOアドベントカレンダーではデータパイプライン及びデータ利活用での非技術的な話を書きましたので、もしよければそちらも見ていただけると嬉しいです。