The data build tool (dbt) is an effective data transformation tool and it supports key AWS analytics services - Redshift, Glue, EMR and Athena. In the previous posts, we discussed benefits of a common data transformation tool and the potential of dbt to cover a wide range of data projects from data warehousing to data lake to data lakehouse. Demo data projects that target Redshift Serverless, Glue, EMR on EC2 and EMR on EKS are illustrated as well. In the last part of the dbt on AWS series, we discuss data transformation pipelines using dbt on Amazon Athena. Subsets of IMDb data are used as source and data models are developed in multiple layers according to the dbt best practices. A list of posts of this series can be found below.

Below shows an overview diagram of the scope of this dbt on AWS series. Athena is highlighted as it is discussed in this post.

Infrastructure

The infrastructure hosting this solution leverages Athena Workgroup, AWS Glue Data Catalog, AWS Glue Crawlers and a S3 bucket. They are deployed using Terraform and the source can be found in the GitHub repository of this post.

Athena Workgroup

The dbt athena adapter requires an Athena workgroup and it only supports the Athena engine version 2. The workgroup used in the dbt project is created as shown below.

 1resource "aws_athena_workgroup" "imdb" {
 2  name = "${local.name}-imdb"
 3
 4  configuration {
 5    enforce_workgroup_configuration    = true
 6    publish_cloudwatch_metrics_enabled = false
 7
 8    engine_version {
 9      selected_engine_version = "Athena engine version 2"
10    }
11
12    result_configuration {
13      output_location = "s3://${local.default_bucket.name}/athena/"
14
15      encryption_configuration {
16        encryption_option = "SSE_S3"
17      }
18    }
19  }
20
21  force_destroy = true
22
23  tags = local.tags
24}

Glue Databases

We have two Glue databases. The source tables and the tables of the staging and intermediate layers are kept in the imdb database. The tables of the marts layer are stored in the _imdb_analytics _database.

 1# athena/infra/main.tf
 2resource "aws_glue_catalog_database" "imdb_db" {
 3  name        = "imdb"
 4  description = "Database that contains IMDb staging/intermediate model datasets"
 5}
 6
 7resource "aws_glue_catalog_database" "imdb_db_marts" {
 8  name        = "imdb_analytics"
 9  description = "Database that contains IMDb marts model datasets"
10}

Glue Crawlers

We use Glue crawlers to create source tables in the imdb database. We can create a single crawler for the seven source tables but it was not satisfactory, especially header detection. Instead a dedicated crawler is created for each of the tables with its own custom classifier where it includes header columns specifically. The Terraform count meta-argument is used to create the crawlers and classifiers recursively.

 1# athena/infra/main.tf
 2resource "aws_glue_crawler" "imdb_crawler" {
 3  count = length(local.glue.tables)
 4
 5  name          = local.glue.tables[count.index].name
 6  database_name = aws_glue_catalog_database.imdb_db.name
 7  role          = aws_iam_role.imdb_crawler.arn
 8  classifiers   = [aws_glue_classifier.imdb_crawler[count.index].id]
 9
10  s3_target {
11    path = "s3://${local.default_bucket.name}/${local.glue.tables[count.index].name}"
12  }
13
14  tags = local.tags
15}
16
17resource "aws_glue_classifier" "imdb_crawler" {
18  count = length(local.glue.tables)
19
20  name = local.glue.tables[count.index].name
21
22  csv_classifier {
23    contains_header = "PRESENT"
24    delimiter       = "\t"
25    header          = local.glue.tables[count.index].header
26  }
27}
28
29# athena/infra/variables.tf
30locals {
31  name        = basename(path.cwd) == "infra" ? basename(dirname(path.cwd)) : basename(path.cwd)
32  region      = data.aws_region.current.name
33  environment = "dev"
34
35  default_bucket = {
36    name = "${local.name}-${data.aws_caller_identity.current.account_id}-${local.region}"
37  }
38
39  glue = {
40    tables = [
41      { name = "name_basics", header = ["nconst", "primaryName", "birthYear", "deathYear", "primaryProfession", "knownForTitles"] },
42      { name = "title_akas", header = ["titleId", "ordering", "title", "region", "language", "types", "attributes", "isOriginalTitle"] },
43      { name = "title_basics", header = ["tconst", "titleType", "primaryTitle", "originalTitle", "isAdult", "startYear", "endYear", "runtimeMinutes", "genres"] },
44      { name = "title_crew", header = ["tconst", "directors", "writers"] },
45      { name = "title_episode", header = ["tconst", "parentTconst", "seasonNumber", "episodeNumber"] },
46      { name = "title_principals", header = ["tconst", "ordering", "nconst", "category", "job", "characters"] },
47      { name = "title_ratings", header = ["tconst", "averageRating", "numVotes"] }
48    ]
49  }
50
51  tags = {
52    Name        = local.name
53    Environment = local.environment
54  }
55}

Project

We build a data transformation pipeline using subsets of IMDb data - seven titles and names related datasets are provided as gzipped, tab-separated-values (TSV) formatted files. This results in three tables that can be used for reporting and analysis.

Save Data to S3

The Axel download accelerator is used to download the data files locally followed by decompressing with the gzip utility. Note that simple retry logic is added as I see download failure from time to time. Finally, the decompressed files are saved into the project S3 bucket using the S3 sync command.

 1# athena/upload-data.sh
 2#!/usr/bin/env bash
 3
 4s3_bucket=$(terraform -chdir=./infra output --raw default_bucket_name)
 5hostname="datasets.imdbws.com"
 6declare -a file_names=(
 7  "name.basics.tsv.gz" \
 8  "title.akas.tsv.gz" \
 9  "title.basics.tsv.gz" \
10  "title.crew.tsv.gz" \
11  "title.episode.tsv.gz" \
12  "title.principals.tsv.gz" \
13  "title.ratings.tsv.gz"
14  )
15
16rm -rf imdb-data
17
18for fn in "${file_names[@]}"
19do
20  download_url="https://$hostname/$fn"
21  prefix=$(echo ${fn::-7} | tr '.' '_')
22  echo "download imdb-data/$prefix/$fn from $download_url"
23  while true;
24  do
25    mkdir -p imdb-data/$prefix
26    axel -n 32 -a -o imdb-data/$prefix/$fn $download_url
27    gzip -d imdb-data/$prefix/$fn
28    num_files=$(ls imdb-data/$prefix | wc -l)
29    if [ $num_files == 1 ]; then
30      break
31    fi
32    rm -rf imdb-data/$prefix
33  done
34done
35
36aws s3 sync ./imdb-data s3://$s3_bucket

Run Glue Crawlers

The Glue crawlers for the seven source tables are executed as shown below.

 1# athena/start-crawlers.sh
 2#!/usr/bin/env bash
 3
 4declare -a crawler_names=(
 5  "name_basics" \
 6  "title_akas" \
 7  "title_basics" \
 8  "title_crew" \
 9  "title_episode" \
10  "title_principals" \
11  "title_ratings"
12  )
13
14for cn in "${crawler_names[@]}"
15do
16  echo "start crawler $cn ..."
17  aws glue start-crawler --name $cn
18done

After the crawlers run successfully, we are able to check the seven source tables. Below shows a query example of one of the source tables in Athena.

Setup dbt Project

We need the dbt-core and dbt-athena-adapter packages for the main data transformation project - the former can be installed as dependency of the latter. The dbt project is initialised while skipping the connection profile as it does not allow to select key connection details such as aws_profile_name and threads. Instead the profile is created manually as shown below. Note that, after this post was complete, it announced a new project repository called dbt-athena-community and new features are planned to be supported in the new project. Those new features cover the Athena engine version 3 and Apache Iceberg support and a new dbt targeting Athena is encouraged to use it.

1$ pip install dbt-athena-adapter
2$ dbt init --skip-profile-setup
3# 09:32:09  Running with dbt=1.3.1
4# Enter a name for your project (letters, digits, underscore): athena_proj

The attributes are self-explanatory, and their details can be checked further in the GitHub repository of the dbt-athena adapter.

 1# athena/set-profile.sh
 2#!/usr/bin/env bash
 3
 4aws_region=$(aws ec2 describe-availability-zones --output text --query 'AvailabilityZones[0].[RegionName]')
 5dbt_s3_location=$(terraform -chdir=./infra output --raw default_bucket_name)
 6dbt_work_group=$(terraform -chdir=./infra output --raw aws_athena_workgroup_name)
 7
 8cat << EOF > ~/.dbt/profiles.yml
 9athena_proj:
10  outputs:
11    dev:
12      type: athena
13      region_name: ${aws_region}
14      s3_staging_dir: "s3://${dbt_s3_location}/dbt/"
15      schema: imdb
16      database: awsdatacatalog
17      work_group: ${dbt_work_group}
18      threads: 3
19      aws_profile_name: <aws-profile>
20  target: dev
21EOF

dbt initialises a project in a folder that matches to the project name and generates project boilerplate as shown below. Some of the main objects are dbt_project.yml, and the model folder. The former is required because dbt doesn’t know if a folder is a dbt project without it. Also it contains information that tells dbt how to operate on the project. The latter is for including dbt models, which is basically a set of SQL select statements. See dbt documentation for more details.

 1$ tree athena/athena_proj/ -L 1
 2athena/athena_proj/
 3├── README.md
 4├── analyses
 5├── dbt_packages
 6├── dbt_project.yml
 7├── logs
 8├── macros
 9├── models
10├── packages.yml
11├── seeds
12├── snapshots
13├── target
14└── tests

We can check Athena connection with the dbt debug command as shown below.

 1$ dbt debug
 209:33:53  Running with dbt=1.3.1
 3dbt version: 1.3.1
 4python version: 3.8.10
 5python path: <path-to-python-path>
 6os info: Linux-5.4.72-microsoft-standard-WSL2-x86_64-with-glibc2.29
 7Using profiles.yml file at /home/<username>/.dbt/profiles.yml
 8Using dbt_project.yml file at <path-to-dbt-project>/dbt_project.yml
 9
10Configuration:
11  profiles.yml file [OK found and valid]
12  dbt_project.yml file [OK found and valid]
13
14Required dependencies:
15 - git [OK found]
16
17Connection:
18  s3_staging_dir: s3://<s3-bucket-name>/dbt/
19  work_group: athena-imdb
20  region_name: ap-southeast-2
21  database: imdb
22  schema: imdb
23  poll_interval: 1.0
24  aws_profile_name: <aws-profile>
25  Connection test: [OK connection ok]
26
27All checks passed!

After initialisation, the model configuration is updated. The project materialisation is specified as view, although it is the default materialisation. Also tags are added to the entire model folder as well as folders of specific layers - staging, intermediate and marts. As shown below, tags can simplify model execution.

 1# athena/athena_proj/dbt_project.yml
 2name: "dbt_glue_proj"
 3
 4...
 5
 6models:
 7  dbt_glue_proj:
 8    +materialized: view
 9    +tags:
10      - "imdb"
11    staging:
12      +tags:
13        - "staging"
14    intermediate:
15      +tags:
16        - "intermediate"
17    marts:
18      +tags:
19        - "marts"

The dbt_utils package is installed for adding tests to the final marts models. The packages can be installed by the dbt deps command.

1# athena/athena_proj/packages.yml
2packages:
3  - package: dbt-labs/dbt_utils
4    version: 0.9.5

Create dbt Models

The models for this post are organised into three layers according to the dbt best practices - staging, intermediate and marts.

Staging

The seven tables that are loaded from S3 are dbt source tables and their details are declared in a YAML file (_imdb_sources.yml). By doing so, we are able to refer to the source tables with the {{ source() }} function. Also we can add tests to source tables. For example below two tests (unique, not_null) are added to the tconst column of the title_basics table below and these tests can be executed by the dbt test command.

 1# athena/athena_proj/models/staging/imdb/_imdb__sources.yml
 2version: 2
 3
 4sources:
 5  - name: imdb
 6    description: Subsets of IMDb data, which are available for access to customers for personal and non-commercial use
 7    tables:
 8      - name: title_basics
 9        description: Table that contains basic information of titles
10        columns:
11          - name: tconst
12            description: alphanumeric unique identifier of the title
13            tests:
14              - unique
15              - not_null
16          - name: titletype
17            description: the type/format of the title (e.g. movie, short, tvseries, tvepisode, video, etc)
18          - name: primarytitle
19            description: the more popular title / the title used by the filmmakers on promotional materials at the point of release
20          - name: originaltitle
21            description: original title, in the original language
22          - name: isadult
23            description: flag that indicates whether it is an adult title or not
24          - name: startyear
25            description: represents the release year of a title. In the case of TV Series, it is the series start year
26          - name: endyear
27            description: TV Series end year. NULL for all other title types
28          - name: runtime minutes
29            description: primary runtime of the title, in minutes
30          - name: genres
31            description: includes up to three genres associated with the title
32      ...

Based on the source tables, staging models are created. They are created as views, which is the project’s default materialisation. In the SQL statements, column names and data types are modified mainly.

 1# athena/athena_proj/models/staging/imdb/stg_imdb__title_basics.sql
 2with source as (
 3
 4    select * from {{ source('imdb', 'title_basics') }}
 5
 6),
 7
 8renamed as (
 9
10    select
11        tconst as title_id,
12        titletype as title_type,
13        primarytitle as primary_title,
14        originaltitle as original_title,
15        cast(isadult as boolean) as is_adult,
16        cast(startyear as int) as start_year,
17        cast(endyear as int) as end_year,
18        cast(runtimeminutes as int) as runtime_minutes,
19        genres
20    from source
21
22)
23
24select * from renamed

Below shows the file tree of the staging models. The staging models can be executed using the dbt run command. As we’ve added tags to the staging layer models, we can limit to execute only this layer by dbt run --select staging.

 1$ tree athena/athena_proj/models/staging/
 2athena/athena_proj/models/staging/
 3└── imdb
 4    ├── _imdb__models.yml
 5    ├── _imdb__sources.yml
 6    ├── stg_imdb__name_basics.sql
 7    ├── stg_imdb__title_akas.sql
 8    ├── stg_imdb__title_basics.sql
 9    ├── stg_imdb__title_crews.sql
10    ├── stg_imdb__title_episodes.sql
11    ├── stg_imdb__title_principals.sql
12    └── stg_imdb__title_ratings.sql

The views in the staging layer can be queried in Athena as shown below.

Intermediate

We can keep intermediate results in this layer so that the models of the final marts layer can be simplified. The source data includes columns where array values are kept as comma separated strings. For example, the genres column of the stg_imdb__title_basics model includes up to three genre values as shown in the previous screenshot. A total of seven columns in three models are columns of comma-separated strings and it is better to flatten them in the intermediate layer. Also, in order to avoid repetition, a dbt macro (f_latten_fields_) is created to share the column-flattening logic.

1# athena/athena_proj/macros/flatten_fields.sql
2{% macro flatten_fields(model, field_name, id_field_name) %}
3    select
4        {{ id_field_name }} as id,
5        field
6    from {{ model }}
7    cross join unnest(split({{ field_name }}, ',')) as x(field)
8{% endmacro %}

The macro function can be added inside a common table expression (CTE) by specifying the relevant model, field name to flatten and ID field name.

 1-- athena/athena_proj/models/intermediate/title/int_genres_flattened_from_title_basics.sql
 2with flattened as (
 3    {{ flatten_fields(ref('stg_imdb__title_basics'), 'genres', 'title_id') }}
 4)
 5
 6select
 7    id as title_id,
 8    field as genre
 9from flattened
10order by id

The intermediate models are also materialised as views, and we can check the array columns are flattened as expected.

Below shows the file tree of the intermediate models. Similar to the staging models, the intermediate models can be executed by dbt run --select intermediate.

 1$ tree athena/athena_proj/models/intermediate/ athena/athena_proj/macros/
 2athena/athena_proj/models/intermediate/
 3├── name
 4│   ├── _int_name__models.yml
 5│   ├── int_known_for_titles_flattened_from_name_basics.sql
 6│   └── int_primary_profession_flattened_from_name_basics.sql
 7└── title
 8    ├── _int_title__models.yml
 9    ├── int_directors_flattened_from_title_crews.sql
10    ├── int_genres_flattened_from_title_basics.sql
11    └── int_writers_flattened_from_title_crews.sql
12
13athena/athena_proj/macros/
14└── flatten_fields.sql

Marts

The models in the marts layer are configured to be materialised as tables in a custom schema. Their materialisation is set to table and the custom schema is specified as analytics while taking _parquet _as the file format. Note that the custom schema name becomes imdb_analytics according to the naming convention of dbt custom schemas. Models of both the staging and intermediate layers are used to create final models to be used for reporting and analytics.

 1-- athena/athena_proj/models/marts/analytics/titles.sql
 2{{
 3    config(
 4        schema='analytics',
 5        materialized='table',
 6        file_format='parquet'
 7    )
 8}}
 9
10with titles as (
11
12    select * from {{ ref('stg_imdb__title_basics') }}
13
14),
15
16principals as (
17
18    select
19        title_id,
20        count(name_id) as num_principals
21    from {{ ref('stg_imdb__title_principals') }}
22    group by title_id
23
24),
25
26names as (
27
28    select
29        title_id,
30        count(name_id) as num_names
31    from {{ ref('int_known_for_titles_flattened_from_name_basics') }}
32    group by title_id
33
34),
35
36ratings as (
37
38    select
39        title_id,
40        average_rating,
41        num_votes
42    from {{ ref('stg_imdb__title_ratings') }}
43
44),
45
46episodes as (
47
48    select
49        parent_title_id,
50        count(title_id) as num_episodes
51    from {{ ref('stg_imdb__title_episodes') }}
52    group by parent_title_id
53
54),
55
56distributions as (
57
58    select
59        title_id,
60        count(title) as num_distributions
61    from {{ ref('stg_imdb__title_akas') }}
62    group by title_id
63
64),
65
66final as (
67
68    select
69        t.title_id,
70        t.title_type,
71        t.primary_title,
72        t.original_title,
73        t.is_adult,
74        t.start_year,
75        t.end_year,
76        t.runtime_minutes,
77        t.genres,
78        p.num_principals,
79        n.num_names,
80        r.average_rating,
81        r.num_votes,
82        e.num_episodes,
83        d.num_distributions
84    from titles as t
85    left join principals as p on t.title_id = p.title_id
86    left join names as n on t.title_id = n.title_id
87    left join ratings as r on t.title_id = r.title_id
88    left join episodes as e on t.title_id = e.parent_title_id
89    left join distributions as d on t.title_id = d.title_id
90
91)
92
93select * from final

The details of the three models can be found in a YAML file (_analytics__models.yml). We can add tests to models and below we see tests of row count matching to their corresponding staging models.

 1# athena/athena_proj/models/marts/analytics/_analytics__models.yml
 2version: 2
 3
 4models:
 5  - name: names
 6    description: Table that contains all names with additional details
 7    tests:
 8      - dbt_utils.equal_rowcount:
 9          compare_model: ref('stg_imdb__name_basics')
10  - name: titles
11    description: Table that contains all titles with additional details
12    tests:
13      - dbt_utils.equal_rowcount:
14          compare_model: ref('stg_imdb__title_basics')
15  - name: genre_titles
16    description: Table that contains basic title details after flattening genres

The models of the marts layer can be tested using the dbt test command as shown below.

 1$ dbt test --select marts
 207:18:42  Running with dbt=1.3.1
 307:18:43  Found 15 models, 17 tests, 0 snapshots, 0 analyses, 473 macros, 0 operations, 0 seed files, 7 sources, 0 exposures, 0 metrics
 407:18:43  
 507:18:48  Concurrency: 3 threads (target='dev')
 607:18:48  
 707:18:48  1 of 2 START test dbt_utils_equal_rowcount_names_ref_stg_imdb__name_basics_ .... [RUN]
 807:18:48  2 of 2 START test dbt_utils_equal_rowcount_titles_ref_stg_imdb__title_basics_ .. [RUN]
 907:18:51  2 of 2 PASS dbt_utils_equal_rowcount_titles_ref_stg_imdb__title_basics_ ........ [PASS in 2.76s]
1007:18:52  1 of 2 PASS dbt_utils_equal_rowcount_names_ref_stg_imdb__name_basics_ .......... [PASS in 3.80s]
1107:18:52  
1207:18:52  Finished running 2 tests in 0 hours 0 minutes and 9.17 seconds (9.17s).
1307:18:52  
1407:18:52  Completed successfully
1507:18:52  
1607:18:52  Done. PASS=2 WARN=0 ERROR=0 SKIP=0 TOTAL=2

Below shows the file tree of the marts models. As with the other layers, the marts models can be executed by dbt run --select marts.

1$ tree athena/athena_proj/models/marts/
2athena/athena_proj/models/marts/
3└── analytics
4    ├── _analytics__models.yml
5    ├── genre_titles.sql
6    ├── names.sql
7    └── titles.sql

Build Dashboard

The models of the marts layer can be consumed by external tools such as Amazon QuickSight. Below shows an example dashboard. The pie chart on the left shows the proportion of titles by genre while the box plot on the right shows the dispersion of average rating by start year.

Generate dbt Documentation

A nice feature of dbt is documentation. It provides information about the project and the data warehouse, and it facilitates consumers as well as other developers to discover and understand the datasets better. We can generate the project documents and start a document server as shown below.

1$ dbt docs generate
2$ dbt docs serve

A very useful element of dbt documentation is data lineage, which provides an overall view about how data is transformed and consumed. Below we can see that the final titles model consumes all title-related stating models and an intermediate model from the name basics staging model.

Summary

In this post, we discussed how to build data transformation pipelines using dbt on AWS Athena. Subsets of IMDb data are used as source and data models are developed in multiple layers according to the dbt best practices. dbt can be used as an effective tool for data transformation in a wide range of data projects from data warehousing to data lake to data lakehouse and it supports key AWS analytics services - Redshift, Glue, EMR and Athena. In this series, we mainly focus on how to develop a data project with dbt targeting variable AWS analytics services. It is quite an effective framework for data transformation and advanced features will be covered in a new series of posts.