社内で RDS, Kintone, Freee など様々なサービスを使っていて、それらの情報をー箇所で集約し、分析するためのデータ基盤の必要がありました。
そのために以下の手順でデータ基盤を構築してみました。

データのまとめ：AWS Glue を用いてデータを S3 に抽出しカタログ化
データ加工：dbt-athena でデータ変換
クエリ実行：Athena でクエリ
データ分析：Metabase でのデータ分析とお可視化

将来的に集めるデータ量が増加見込みなので、コスト効率と拡張性の面で S3 での保存は適切かと考えました。
また、開発チュー自体少人数なので、煩雑な管理作業を避けるためにサーバーレスや自動化に注力し、ETL（Extract, Transform, Load）ワークフローは AWS Glue、Athena、dbt で構築しました。

安全かつスマートにインフラを管理

インフラ管理には AWS コンソールで直接操作するのではなく、CloudFormation を活用していますので、CloudFormation と SaaS サービスの連携例も記載しています。

全体図

1. データのまとめ: Glue を使ってデータを S3 に格納

<> の囲んだものは適切な設定で置き換えてください。

先に S3 Bucket と Glue Job で利用するロールを作成します。

S3DataLakeBucket: # 今回はこのBucketにはデータ以外に、Glue Job の ファイルや他のライブラリも一緒におきます
  Type: AWS::S3::Bucket
  Properties:
    BucketName: <BUCKET_NAME>
    PublicAccessBlockConfiguration:
      BlockPublicAcls: true
      BlockPublicPolicy: true
      IgnorePublicAcls: true
      RestrictPublicBuckets: true
    VersioningConfiguration:
      Status: Enabled
ETLGlueJobRole: # Glue Job が S3やRDS にアクセスするための Roleです。
  Type: AWS::IAM::Role
  Properties:
    RoleName: etl-glue-job-role
    AssumeRolePolicyDocument:
      Version: '2012-10-17'
      Statement:
        - Effect: Allow
          Principal:
            Service: glue.amazonaws.com
          Action: sts:AssumeRole
    Policies:
      - PolicyName: MyGlueJobPolicy # サンプルのため一部のPolicyは全てのリソースに対して権限を付与
        PolicyDocument:
          Version: '2012-10-17'
          Statement:
            - Effect: Allow
              Action:
                - logs:CreateLogGroup
                - logs:CreateLogStream
                - logs:PutLogEvents
              Resource: arn:aws:logs:*:*:*
            - Effect: Allow
              Action:
                - ssm:GetParameter
              Resource: '*'
            - Effect: Allow
              Action:
                - s3:GetObject
                - s3:ListBucket
                - s3:GetBucketLocation
                - s3:PutObject
                - s3:DeleteObject
              Resource:
                - !GetAtt S3DataLakeBucket.Arn
                - !Join
                  - ''
                  - - !GetAtt S3DataLakeBucket.Arn
                    - /*
            - Effect: Allow
              Action:
                - glue:StartJobRun
              Resource: arn:aws:glue:*:*:job/*
            - Sid: ForGlueCrawler
              Effect: Allow
              Action:
                - glue:*
              Resource:
                - arn:aws:glue:ap-northeast-1:*:catalog
                - arn:aws:glue:ap-northeast-1:*:connection/*
                - arn:aws:glue:ap-northeast-1:<AWS_ACCOUNT_ID>:database/*
                - arn:aws:glue:ap-northeast-1:<AWS_ACCOUNT_ID>:table/*
            - Effect: Allow
              Action:
                - secretsmanager:GetResourcePolicy
                - secretsmanager:GetSecretValue
                - secretsmanager:DescribeSecret
                - secretsmanager:ListSecretVersionIds
              Resource:
                - arn:aws:secretsmanager:ap-northeast-1:<AWS_ACCOUNT_ID>:secret:*
            - Action:
                - ec2:DescribeVpcEndpoints
                - ec2:DescribeRouteTables
                - ec2:DescribeNetworkInterfaces
                - ec2:CreateNetworkInterface
                - ec2:DeleteNetworkInterface
                - ec2:DescribeSecurityGroups
                - ec2:DescribeSubnets
                - ec2:DescribeVpcAttribute
              Resource: '*'
              Effect: Allow
            - Effect: Allow
              Action:
                - ec2:CreateTags
                - ec2:DeleteTags
              Condition:
                ForAllValues:StringEquals:
                  aws:TagKeys:
                    - aws-glue-service-resource
              Resource:
                - arn:aws:ec2:*:*:network-interface/*
                - arn:aws:ec2:*:*:security-group/*
                - arn:aws:ec2:*:*:instance/*
            - Effect: Allow
              Action:
                - athena:StartQueryExecution
                - athena:GetQueryExecution
                - athena:GetQueryResults
              Resource: '*'

RDS のデータを定期的に S3 にエクスポート

RDS のデータは Glue ETL を使って全てのデータを S3 に出力しています。
今回は MySQL のデータを S3 に Parquet として出力する template を例として見てみます。

RDS に接続するための Glue Connection を作成

SampleDBConnectionSecurityGroup:
  Type: AWS::EC2::SecurityGroup
  Properties:
    GroupDescription: 'for db connection'
    GroupName: sgp-glue-connection
    VpcId: <VPC_ID>
    SecurityGroupEgress:
      - IpProtocol: '-1'
        CidrIp: 0.0.0.0/0
      - IpProtocol: '-1'
        CidrIpv6: '::/0'
SampleDBConnectionSecurityGroupIngress:
  Type: AWS::EC2::SecurityGroupIngress
  Properties:
    GroupId: !Ref SampleDBConnectionSecurityGroup
    IpProtocol: tcp
    FromPort: '0'
    ToPort: '65535'
    SourceSecurityGroupId: !Ref SampleDBConnectionSecurityGroup
SampleDBConnection:
  Type: AWS::Glue::Connection
  Properties:
    CatalogId: !Ref AWS::AccountId
    ConnectionInput:
      ConnectionType: JDBC
      ConnectionProperties:
        JDBC_CONNECTION_URL: 'jdbc:mysql://<DB_HOST>:<DB_PORT>/<DB_NAME>' # mysqlの場合
        SECRET_ID: <SECRET_ID> # パスワードは SecretManger で保存されているので、SecretManager の IDで指定する - https://docs.aws.amazon.com/ja_jp/AWSCloudFormation/latest/UserGuide/aws-properties-glue-connection-connectioninput.html#cfn-glue-connection-connectioninput-connectionproperties
      Name: glue-connection
      PhysicalConnectionRequirements:
        SecurityGroupIdList:
          - !Ref SampleDBConnectionSecurityGroup
        SubnetId: <SUBNET_ID>

SecurityGroup は自己参照ルールを追加する必要があります

CloudFormation で Glue Connection を作成すると成功にはなりますが、GlueJobでその connection を利用しようとすると Unable to resolve any valid connection になる場合がありましたので、そうなった場合は AWS コンソールで対象 connection を選択→編集→何も変更せず保存　をするとちゃんと接続できるようになります。

次にGlue Data Catalog で Database を作成します。
履歴データも保存しておきたいので履歴データは別 DB として保存しています。
partition で最新データを管理するのも可能ですが、SQL で最新データを取得時にいちいち partition キーを指定したくないため、別データベースで管理した方が良さそうな気がしたので、別々で管理しています。

SampleDBDatabase:
  Type: AWS::Glue::Database
  Properties:
    CatalogId: !Ref AWS::AccountId
    DatabaseInput:
      Name: sample-db
      Description: 'RDS'
SampleDBHistoryDatabase:
  Type: AWS::Glue::Database
  Properties:
    CatalogId: !Ref AWS::AccountId
    DatabaseInput:
      Name: sample-db-history
      Description: 'DB_HISTORY'

s3://<BUCKET_NAME>/glueJobs/ に以下のファイル2つをアップロードします。

SampleDBGlueJob.py（このファイルはGlue Studio の Visual ETL をベースに全テーブルを出力する形にしています）

import datetime
import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from utils_s3 import get_s3_history_prefix, get_s3_latest_prefix

args = getResolvedOptions(
    sys.argv, ["JOB_NAME", "s3Bucket", "s3Folder", "dbConnection"]
)
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args["JOB_NAME"], args)

today_string = datetime.datetime.today().strftime("%Y%m%d")

# get all tables from app db
allTables = glueContext.create_dynamic_frame.from_options(
    connection_type="mysql",
    connection_options={
        "useConnectionProperties": "true",
        "dbtable": "information_schema.tables",
        "connectionName": args["dbConnection"],
    },
    transformation_ctx="MySQLtable_node1",
)
targetTables = (
    allTables.toDF()
    .filter("TABLE_SCHEMA != 'information_schema'")
    .filter("TABLE_TYPE = 'BASE TABLE'")
    .collect()
)

for row in targetTables:
    table_name = row["TABLE_NAME"]
    schema_name = row["TABLE_SCHEMA"]
    print(f"SCHEMA_NAME: {schema_name}, TABLE_NAME: {table_name}")

    ds = glueContext.create_dynamic_frame.from_options(
        connection_type="mysql",
        connection_options={
            "useConnectionProperties": "true",
            "dbtable": table_name,
            "connectionName": args["dbConnection"],
        },
        transformation_ctx="MySQLtable_node1",
    )

    # 最新情報として保存
    latestDataPath = (
        f's3://{args["s3Bucket"]}/{get_s3_latest_prefix(args["s3Folder"], table_name)}/'
    )
    glueContext.purge_s3_path(latestDataPath, {"retentionPeriod": 0})
    glueContext.write_dynamic_frame.from_options(
        frame=ds,
        connection_type="s3",
        connection_options={"path": latestDataPath},
        format="parquet",
    )

    # 履歴情報として保存
    historyDataPath = f's3://{args["s3Bucket"]}/{get_s3_history_prefix(args["s3Folder"], table_name)}/'
    glueContext.purge_s3_path(historyDataPath, {"retentionPeriod": 0})

    source_dyf = glueContext.create_dynamic_frame_from_options(
        connection_type="s3",
        connection_options={"path": latestDataPath, "recurse": True},
        format="parquet",
    )

    glueContext.write_dynamic_frame.from_options(
        frame=source_dyf,
        connection_type="s3",
        connection_options={"path": historyDataPath, "overwrite": True},
        format="parquet",
    )

job.commit()

utils_s3.py も S3にアップロード

import datetime
import pytz
import boto3

tz = pytz.timezone('Asia/Tokyo')
now = datetime.datetime.now(tz)
history_year = now.strftime('%Y')
history_month = now.strftime('%m')
history_day = now.strftime('%d')

s3 = boto3.client("s3")

def get_s3_latest_prefix(prefix, tableName):
    return f'{prefix}/latest/{tableName}';

def get_s3_history_prefix(prefix,tableName):
    return f'{prefix}/history/{tableName}/history_year={history_year}/history_month={history_month}/history_day={history_day}';

def save_local_file_to_s3(bucket, prefix, tableName, fileName):
    """
    ローカルファイルをS3にアップロードする(履歴も保存される)
    - s3://{bucket}/{prefix}/latest/{tableName}/{fileName}
    - s3://{bucket}/{prefix}/history/{tableName}/history_year=YYYY/history_month=MM/history_day=DD/{fileName}
    """
    s3_latest_path = f'{get_s3_latest_prefix(prefix, tableName)}/{fileName}'
    s3.upload_file(fileName, bucket, s3_latest_path)

    s3_history_path = f'{get_s3_history_prefix(prefix, tableName)}/{fileName}'
    s3.copy_object(
        Bucket=bucket,
        CopySource={'Bucket': bucket, 'Key': s3_latest_path},
        Key=s3_history_path
    )

次に、Workflow を作成し、Workflow で Glue Triggers、Glue Crawlers などが自動で発火されるようにします。

SampleDBETLWorkFlow:
  Type: AWS::Glue::Workflow
  Properties:
    Name: sample-db-etl-workflow
SampleDBGlueJob: # RDS のデータを抽出する Glue Job です
  Type: AWS::Glue::Job
  Properties:
    Name: sample-db-glue-job
    Role: !GetAtt ETLGlueJobRole.Arn
    Command:
      Name: glueetl
      ScriptLocation: !Join
        - ''
        - - s3://
          - !Ref S3DataLakeBucket
          - /glueJobs/SampleDBGlueJob.py
    GlueVersion: '4.0'
    WorkerType: G.1X
    NumberOfWorkers: 2
    Connections:
      Connections:
        - !Ref SampleDBConnection
    DefaultArguments:
      '--s3Bucket': !Ref S3DataLakeBucket
      '--s3Folder': bronze/sample-db
      '--dbConnection': !Ref SampleDBConnection
      '--extra-py-files': !Join
        - ''
        - - s3://
          - !Ref S3DataLakeBucket
          - /glueJobs/utils_s3.py
SampleDBGlueJobTrigger: # 毎日出力するための Trigger
  Type: AWS::Glue::Trigger
  Properties:
    Actions:
      - JobName: !Ref SampleDBGlueJob
    Name: sample-db-glue-job-trigger
    Schedule: cron(0 * * * ? *)
    Type: SCHEDULED
    WorkflowName: !Ref SampleDBETLWorkFlow
    StartOnCreation: true
SampleDBCrawler: # 出力されたファイルを Glue Data Catalog に反映するためには、Glue Crawler を使います
  Type: AWS::Glue::Crawler
  Properties:
    DatabaseName: !Ref SampleDBDatabase
    Role: !GetAtt ETLGlueJobRole.Arn
    Targets:
      S3Targets:
        - Path: !Join
            - ''
            - - s3://
              - !Ref S3DataLakeBucket
              - /bronze/sample-db/latest/
    Name: sample-db-crawler
    Configuration: '{"Version":1,"Grouping":{"TableLevelConfiguration":5}}' # 基本こちら設定なくても Glue Crawler の方で階層を自動で判断してくれますが、念の為設定しています
SampleDBCrawlerTrigger:
  Type: AWS::Glue::Trigger
  Properties:
    Actions:
      - CrawlerName: !Ref SampleDBCrawler
    Name: sample-db-crawler-trigger
    Predicate:
      Conditions:
        - LogicalOperator: EQUALS
          JobName: !Ref SampleDBGlueJob
          State: SUCCEEDED # GlueJob のデータ抽出が成功した場合にのみ Crawler を発火させます
      Logical: ANY
    Type: CONDITIONAL
    WorkflowName: !Ref SampleDBETLWorkFlow
    StartOnCreation: true
SampleDBHistoryCrawler:
  Type: AWS::Glue::Crawler
  Properties:
    DatabaseName: !Ref SampleDBHistoryDatabase
    Role: !GetAtt ETLGlueJobRole.Arn
    Targets:
      S3Targets:
        - Path: !Join
            - ''
            - - s3://
              - !Ref S3DataLakeBucket
              - /bronze/sample-db/history/
    Name: sample-db-history-crawler
    Configuration: '{"Version":1,"Grouping":{"TableLevelConfiguration":5,"TableGroupingPolicy":"CombineCompatibleSchemas"},"CreatePartitionIndex":true}' # サンプルDBは 日付で Partitionします
    RecrawlPolicy:
      RecrawlBehavior: CRAWL_NEW_FOLDERS_ONLY # 追加のpartition分だけクロールするように設定
SampleDBHistoryCrawlerTrigger:
  Type: AWS::Glue::Trigger
  Properties:
    Actions:
      - CrawlerName: !Ref SampleDBHistoryCrawler
    Name: sample-db-history-crawler-trigger
    Predicate:
      Conditions:
        - LogicalOperator: EQUALS
          JobName: !Ref SampleDBGlueJob
          State: SUCCEEDED
      Logical: ANY
    Type: CONDITIONAL
    WorkflowName: !Ref SampleDBETLWorkFlow
    StartOnCreation: true

上記でRDSのデータが毎日 S3 に出力されます。
WorkFlow でステータス確認

Athenaからもクエリすることができます。

社内では簡易データベースなども EC2 で動いており、それらも上記と同じ方法で S3 に出力を行っています。

Kintone のデータを S3 に出力

Kintone データの出力は Glue Python Shell で行います。
Kintone API を叩いて S3 に格納する形です。
GlueJob 以外のリソースは上記の設定とかわりません。

KintoneGlueJob:
  Type: AWS::Glue::Job
  Properties:
    Name: kintone-glue-job
    Role: !GetAtt ETLGlueJobRole.Arn
    Command:
      Name: pythonshell
      PythonVersion: '3.9'
      ScriptLocation: !Join
        - ''
        - - s3://
          - !Ref S3DataWarehouse
          - /glueJobs/KintoneGlueJob.py
    DefaultArguments:
      '--s3Bucket': !Ref S3DataWarehouse
      '--s3Folder': bronze/kintone

      '--extra-py-files': !Join
        - ''
        - - s3://
          - !Ref S3DataWarehouse
          - /glueJobs/utils_s3.py

KintoneGlueJob.py

import json
import os
import sys
import boto3
import datetime
import requests
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
from awsglue.utils import getResolvedOptions
import logging
from dateutil.parser import parse
from utils_s3 import save_local_file_to_s3

s3 = boto3.resource("s3")
ssm = boto3.client("ssm")

args = getResolvedOptions(
    sys.argv,
    ["stage", "s3Bucket", "s3Folder", "ssmKintoneAPIUrlName", "ssmKintoneAPIKeysName"],
)

KINTONE_DOMAIN = ssm.get_parameter(Name=args["ssmKintoneAPIUrlName"])["Parameter"][
    "Value"
]
KINTONE_API_KEYS = json.loads(
    ssm.get_parameter(Name=args["ssmKintoneAPIKeysName"], WithDecryption=True)[
        "Parameter"
    ]["Value"]
)

# 設定自体は DynamoDB に保存した方が良さそう
KINTONE_SYNC_LIST = [
    {
        "tableNameForGlue": "kintone_app",
        "kintoneAppId": "0",
        "kintoneQuery": "order by $id asc",
        "description": "KINTONEテストアプリ",
        "kintoneFields": [
            {
                "field_code": "id",
                "label": "id",
            },
            {
                "field_code": "Name",
                "label": "Name",
            },
            {
                "field_code": "Age",
                "label": "Age",
                "type": "NUMBER",
            },
        ],
    }
]

logger = logging.getLogger()
logger.setLevel(logging.INFO)

# glue で logger を動くようにする
logger.info = lambda msg: print(f"{datetime.datetime.now()}	INFO	{msg}")

errors = []

for sync in KINTONE_SYNC_LIST:
    logger.info(f'start - {sync["tableNameForGlue"]}')

    # cursor 作成
    url = f"{KINTONE_DOMAIN}/k/v1/records/cursor.json"
    headers = {
        "X-Cybozu-API-Token": f"{KINTONE_API_KEYS[sync['kintoneAppId']]}",
        "Content-Type": "application/json",
    }
    fieldCodeList = ["$id"] + [
        field["field_code"] for field in (sync["kintoneFields"] or [])
    ]
    response = requests.post(
        url,
        headers=headers,
        json={
            "app": sync["kintoneAppId"],
            "fields": fieldCodeList,
            "query": sync["kintoneQuery"],
            "size": 500,
        },
    )
    response.raise_for_status()
    cursor = response.json()
    logger.info(cursor)

    writer = None
    fileName = f'{sync["tableNameForGlue"]}.parquet'
    retrieved_records_total = 0

    try:
        while True:
            # cursor からレコード取得
            response = requests.get(url, headers=headers, json=cursor)
            response.raise_for_status()
            result = response.json()

            retrieved_records_total += len(result["records"])
            logger.info(
                {
                    "retrieved_records_total": retrieved_records_total,
                    "next": result["next"],
                }
            )

            schema = pa.schema([], metadata={"description": "test description"})

            # 取得した分だけ parquetに書き込む
            convertedRecordList = []
            for record in result["records"]:
                convertedRecord = {}
                schema = pa.schema([])

                # sort record property by fieldCode
                sorted_items = sorted(
                    record.items(),
                    key=lambda key_value: (
                        fieldCodeList.index(key_value[0])
                        if key_value[0] in fieldCodeList
                        else len(fieldCodeList) + 1
                    ),
                )

                # loop through fields
                for originalFieldCode, typeObj in sorted_items:
                    customField = next(
                        (
                            field
                            for field in sync["kintoneFields"]
                            if field["field_code"] == originalFieldCode
                        ),
                        {},
                    )

                    fieldCode = customField.get("label") or originalFieldCode
                    fieldType = customField.get("type") or typeObj["type"]
                    value = typeObj["value"]
                    try:
                        if fieldType in ["__ID__", "NUMBER"]:
                            convertedRecord[fieldCode] = int(float(value or 0))
                            schema = schema.append(pa.field(fieldCode, pa.int64()))
                        elif fieldType in ["DOUBLE"]:
                            convertedRecord[fieldCode] = float(value or 0)
                            schema = schema.append(pa.field(fieldCode, pa.float64()))
                        elif fieldType in ["DATE"]:
                            convertedRecord[fieldCode] = (
                                None
                                if value is None or value == ""
                                else parse(value).date()
                            )
                            schema = schema.append(pa.field(fieldCode, pa.date32()))
                        elif fieldType in ["DATETIME", "CREATED_TIME", "UPDATED_TIME"]:
                            convertedRecord[fieldCode] = (
                                None
                                if value is None or value == ""
                                else datetime.datetime.strptime(
                                    value, "%Y-%m-%dT%H:%M:%SZ"
                                )
                            )
                            schema = schema.append(
                                pa.field(fieldCode, pa.timestamp("ns"))
                            )
                        elif fieldType in ["TIME"]:
                            # Athena does not support TIME type, so convert to TIMESTAMP
                            convertedRecord[fieldCode] = (
                                None
                                if value is None or value == ""
                                else datetime.datetime.utcfromtimestamp(0).strptime(
                                    value, "%H:%M"
                                )
                            )
                            schema = schema.append(
                                pa.field(fieldCode, pa.timestamp("ns"))
                            )
                        elif fieldType in ["SUBTABLE"]:
                            print("--------------------")
                            print(fieldType)
                            convertedRecord[fieldCode] = (
                                None
                                if value is None or value == ""
                                else json.dumps(value)
                            )
                            schema = schema.append(pa.field(fieldCode, pa.string()))
                        else:
                            convertedRecord[fieldCode] = (
                                None if value is None or value == "" else str(value)
                            )
                            schema = schema.append(pa.field(fieldCode, pa.string()))
                    except Exception as e:
                        logger.info(record)
                        logger.error(
                            f"originalFieldCode: {originalFieldCode}, fieldType: {fieldType}, value: {value}"
                        )
                        raise e

                convertedRecordList.append(convertedRecord)

            if writer is None:
                writer = pq.ParquetWriter(fileName, schema)

            df = pd.DataFrame(convertedRecordList)
            writer.write_table(pa.Table.from_pandas(df, schema=schema))

            if result["next"] == False:
                break

        writer.close()

        # upload file to s3
        save_local_file_to_s3(
            args["s3Bucket"], args["s3Folder"], sync["tableNameForGlue"], fileName
        )

        logger.info(f'end - {sync["tableNameForGlue"]}')
    except Exception as e:
        if writer is not None:
            writer.close()

        errors.append(e)
        logger.info(f'errorOccurred - {sync["tableNameForGlue"]}')
    finally:
        # delete cursor
        requests.delete(url, headers=headers, json=cursor)
        logger.info("cursor deleted")

        # delete local file
        os.remove(fileName)
        logger.info(f"local file deleted")

# log errors
if len(errors) > 0:
    for error in errors:
        logger.error(error)

Freee

Freee も Kintone 同様で Freee API を叩いて結果を S3 に格納しています

2. データ加工：dbt-athena でデータ変換

glue job で dbt を設定し dbt でデータの処理や変換を行います

DbtGlueJob:
  Type: AWS::Glue::Job
  Properties:
    Name: dbt-glue-job
    Role: !GetAtt ETLGlueJobRole.Arn
    Command:
      Name: pythonshell
      PythonVersion: '3.9'
      ScriptLocation: !Join
        - ''
        - - s3://
          - !Ref S3DataLakeBucket
          - /glueJobs/dbt.py
    ExecutionProperty:
      MaxConcurrentRuns: 4
    DefaultArguments:
      '--additional-python-modules': dbt-athena-community
      '--s3Bucket': !Ref S3DataLakeBucket
      '--dbtRepositoryZipKey': glue/dbt/my_dbt_project.zip # dbt projectの設定は https://docs.getdbt.com/docs/introduction 参照
      '--models': '*'
DbtTrigger:
  Type: AWS::Glue::Trigger
  Properties:
    Actions:
      - JobName: !Ref DbtGlueJob
        Arguments:
          '--models': sample-db
    Name: dbt-trigger
    Predicate:
      Conditions:
        - LogicalOperator: EQUALS
          JobName: !Ref SampleDBGlueJob
          State: SUCCEEDED
      Logical: ANY
    Type: CONDITIONAL
    WorkflowName: !Ref SampleDBETLWorkFlow
    StartOnCreation: true

dbt.py

import subprocess
import sys
import boto3
from dbt.cli.main import dbtRunner
from awsglue.utils import getResolvedOptions
import os

args = getResolvedOptions(
    sys.argv, ["s3Bucket", "dbtRepositoryZipKey", "models"],
)

s3 = boto3.resource('s3')
s3.Bucket(args["s3Bucket"]).download_file(Filename="dbt.zip", Key=args["dbtRepositoryZipKey"])

cmd = r"unzip dbt.zip"
print(subprocess.check_call(cmd.split()))

dbt = dbtRunner()
dbtArgs = ['run']

silverPrefix = f'silver.{args["models"]}';
goldPrefix = f'gold.{args["models"]}';

print(f'running {silverPrefix} models')
dbtArgs.append('--models')
dbtArgs.append(silverPrefix)
dbtRunnerResult = dbt.invoke(dbtArgs)
print('silver completed')

print(f'running {goldPrefix} models')
dbtArgs.pop()
dbtArgs.append(f'gold.{args["models"]}')
dbtRunnerResult = dbt.invoke(dbtArgs)
print('gold completed')

dbt project で SQL を書いて bronze から silver や gold にデータ変換を行う感じです。

上記で、Workflow 自体は以下のような感じで動きます。

3.クエリ実行：Athena でクエリ

GlueJob で出力したデータや dbt で変換されたデータは Athena でクエリしています

4. データ分析：Metabase でのデータ分析とお可視化

現状 metabase を使ってデータ分析や・可視化を行っています。
（ただ、metabase もサーバの管理が必要なのと、いくつかのグラフで表示速度が最適ではないため、今後は QuickSight に移行するのも検討している状況です。）

まとめ

多様なデータソースの集約
- 今まで、AWS RDS, Kintone, Freee など、バラバラに散乱していいるデータを一元管理することで、ビジネスインサイトなの精度向上や意思決定の迅速化に繋ぐのではないかと思っています。
サーバレスアーキテクチャの採用
- AWS Glue と Athena を使用してサーバーレスな ETL ワークフローが実現できました
- 少人数の開発チームでも運用が容易であり、コスト効率と拡張性を重視したインフラ設計になっていると思います
- 今後は、GA や Google Search Console のデータも AWS AppFlow 経由で管理する形で、より容易は構築と正確なインサイトが得られるようにしていこうと考えております。