advanced⏱70 minLección 6 de 11

Advanced Jinja2 Macros and Custom Materializations

Write production-grade Jinja2 macros for DRY SQL patterns, build reusable macro libraries, and create custom materializations for Redshift-specific patterns like append-only event logs and external table refresh.

Advanced Jinja2 Macros and Custom Materializations

Jinja2 is dbt's template engine — it transforms SQL from a static query language into a dynamic, composable one. Mastering Jinja2 is the difference between a project that scales to 500 models with a small team and one that becomes unmaintainable. This module covers production-grade macro patterns and custom materializations for Redshift-specific use cases.

Jinja2 in dbt: What's Available

dbt extends Jinja2 with its own functions and objects. The three layers are:

┌─────────────────────────────────────┐
│  dbt-specific functions              │
│  ref(), source(), config(), this     │
│  run_query(), log(), execute        │
├─────────────────────────────────────┤
│  Standard Jinja2                     │
│  {% if %}, {% for %}, {% set %}      │
│  filters, tests, whitespace control  │
├─────────────────────────────────────┤
│  Python built-ins exposed by Jinja2  │
│  range(), namespace(), loop.*        │
└─────────────────────────────────────┘

Macro Fundamentals: Beyond the Basics

Variable Scoping with namespace()

Jinja2 loop variables don't update outer scope without namespace():

sql

-- macros/utils/string_utils.sql

-- WRONG: result is always empty string after the loop
{% macro join_columns_wrong(columns) %}
    {% set result = '' %}
    {% for col in columns %}
        {% set result = result ~ col %}  {# This does NOT update outer result #}
    {% endfor %}
    {{ result }}
{% endmacro %}

-- CORRECT: use namespace() for mutable state inside loops
{% macro join_columns(columns, separator=', ') %}
    {% set ns = namespace(parts=[]) %}
    {% for col in columns %}
        {% set ns.parts = ns.parts + [col] %}
    {% endfor %}
    {{ ns.parts | join(separator) }}
{% endmacro %}

Returning Values from Macros

sql

-- macros/utils/get_column_names.sql
{% macro get_column_names(relation) %}
    {% set query %}
        select column_name
        from information_schema.columns
        where table_schema = '{{ relation.schema }}'
          and table_name   = '{{ relation.name }}'
        order by ordinal_position
    {% endset %}

    {% set results = run_query(query) %}

    {% if execute %}
        {% set columns = results.columns[0].values() %}
        {{ return(columns) }}
    {% else %}
        {{ return([]) }}
    {% endif %}
{% endmacro %}

Using the returned value:

sql

-- models/marts/facts/fct_orders.sql
{% set cols = get_column_names(ref('stg_orders')) %}
select
    {% for col in cols %}
        {{ col }}{% if not loop.last %},{% endif %}
    {% endfor %}
from {{ ref('stg_orders') }}

Production Macro Library for Redshift

1. Dynamic Union of Multiple Sources

sql

-- macros/redshift/union_relations.sql
{% macro union_relations(relations, col_names=none) %}
    {%- for relation in relations -%}
        select
            {%- if col_names is not none -%}
                {%- for col in col_names %}
            {{ col }}{% if not loop.last %},{% endif %}
                {%- endfor %}
            {%- else -%}
            *
            {%- endif %}
        from {{ relation }}
        {% if not loop.last %}union all{% endif %}
    {%- endfor -%}
{% endmacro %}

Usage:

sql

-- models/marts/fct_all_events.sql
{{ union_relations([
    ref('fct_web_events'),
    ref('fct_mobile_events'),
    ref('fct_api_events')
]) }}

2. Pivot Macro (Redshift-safe)

Redshift does not have a native PIVOT. This macro generates the CASE/SUM pattern:

sql

-- macros/redshift/pivot.sql
{% macro pivot(
    column,
    values,
    alias=true,
    agg='sum',
    then_value=1,
    else_value=0,
    quote_identifiers=false,
    distinct=false
) %}
    {%- for value in values %}
    {{ agg }}(
        {%- if distinct %} distinct {% endif -%}
        case
            when {{ column }} = '{{ value }}'
            then {{ then_value }}
            else {{ else_value }}
        end
    )
    {%- if alias %} as {% if quote_identifiers %}"{{ value }}"{% else %}{{ value | replace(' ', '_') | lower }}{% endif %}{% endif %}
    {%- if not loop.last %},{%- endif %}
    {%- endfor %}
{% endmacro %}

sql

-- models/marts/fct_order_status_pivot.sql
select
    report_date,
    {{ pivot(
        column='status',
        values=['Pending', 'Shipped', 'Cancelled', 'Rejected'],
        agg='count'
    ) }}
from {{ ref('fct_orders') }}
group by 1

3. Generate Date Spine

sql

-- macros/redshift/date_spine.sql
{% macro date_spine(start_date, end_date, datepart='day') %}
    with date_series as (
        select
            (
                '{{ start_date }}'::date
                + generate_series(
                    0,
                    datediff(
                        '{{ datepart }}',
                        '{{ start_date }}'::date,
                        '{{ end_date }}'::date
                    )
                ) * interval '1 {{ datepart }}'
            )::date as date_day
    )
    select date_day from date_series
{% endmacro %}

sql

-- models/marts/dim_date.sql
{{ config(materialized='table', dist='all', sort='date_day') }}

{{ date_spine(
    start_date='2020-01-01',
    end_date='2030-12-31',
    datepart='day'
) }}

4. Redshift UNLOAD to S3 Macro

sql

-- macros/redshift/unload_to_s3.sql
{% macro unload_to_s3(
    relation,
    s3_path,
    iam_role,
    format='PARQUET',
    partition_by=none,
    max_file_size='6.2 GB',
    parallel=true
) %}
    {% set query %}
        unload (
            'select * from {{ relation }}'
        )
        to '{{ s3_path }}'
        iam_role '{{ iam_role }}'
        format {{ format }}
        {% if partition_by %}
        partition by ({{ partition_by | join(', ') }})
        {% endif %}
        allowoverwrite
        maxfilesize '{{ max_file_size }}'
        {% if parallel %}parallel on{% else %}parallel off{% endif %};
    {% endset %}

    {% if execute %}
        {% do run_query(query) %}
        {{ log("Unloaded " ~ relation ~ " to " ~ s3_path, info=true) }}
    {% endif %}
{% endmacro %}

Usage as an operation:

bash

dbt run-operation unload_to_s3 --args "{
    relation: 'analytics.marts.fct_orders',
    s3_path: 's3://my-data-lake/exports/fct_orders/',
    iam_role: 'arn:aws:iam::123456789012:role/RedshiftS3Role',
    format: 'PARQUET',
    partition_by: ['order_date']
}"

5. Adaptive Sort Key Recommendation Macro

sql

-- macros/redshift/recommend_sort_keys.sql
{% macro recommend_sort_keys(schema, table, limit=10) %}
    {% set query %}
        select
            schemaname,
            tablename,
            "column"    as column_name,
            usename     as query_count,
            plannode
        from svl_qlog
        join pg_user using (usesysid)
        where schemaname = '{{ schema }}'
          and tablename  = '{{ table }}'
        order by query_count desc
        limit {{ limit }}
    {% endset %}

    {% if execute %}
        {% set results = run_query(query) %}
        {% for row in results.rows %}
            {{ log(row, info=true) }}
        {% endfor %}
    {% endif %}
{% endmacro %}

Custom Materializations

Custom materializations let you define new persistence patterns beyond dbt's built-ins. They are Jinja2 blocks stored in macros/materializations/.

Anatomy of a Custom Materialization

sql

-- macros/materializations/append_only_table.sql
{% materialization append_only_table, adapter='redshift' %}

    {# Required configuration #}
    {%- set target_relation = this.incorporate(type='table') -%}

    {# Get the run adapter #}
    {%- set existing_relation = load_cached_relation(this) -%}

    {# Run pre-hooks #}
    {{ run_hooks(pre_hooks) }}

    {# Create table on first run #}
    {% if existing_relation is none %}
        {% call statement('main') %}
            {{ create_table_as(false, target_relation, sql) }}
        {% endcall %}

    {# Append on subsequent runs #}
    {% else %}
        {% call statement('main') %}
            insert into {{ target_relation }}
            ({{ sql }})
        {% endcall %}
    {% endif %}

    {# Run post-hooks #}
    {{ run_hooks(post_hooks) }}

    {# Update the relation cache #}
    {{ return({'relations': [target_relation]}) }}

{% endmaterialization %}

Using the custom materialization:

sql

-- models/raw/raw_click_stream.sql
{{ config(
    materialized='append_only_table',
    dist='session_id',
    sort=['event_timestamp'],
    sort_type='compound',
    backup=false
) }}

select
    session_id,
    user_id,
    event_timestamp,
    page_url,
    referrer
from {{ source('raw', 'click_stream') }}
where loaded_at > (
    select coalesce(
        max(event_timestamp)::timestamp,
        '1970-01-01'::timestamp
    )
    from {{ this }}
)

Custom Materialization: Redshift External Table Refresh

For Redshift Spectrum external tables that need periodic metadata refresh:

sql

-- macros/materializations/spectrum_external_table.sql
{% materialization spectrum_external_table, adapter='redshift' %}

    {%- set external_schema = config.require('external_schema') -%}
    {%- set s3_location     = config.require('s3_location') -%}
    {%- set file_format     = config.get('file_format', 'parquet') -%}
    {%- set partition_cols  = config.get('partition_cols', []) -%}
    {%- set table_properties = config.get('table_properties', {}) -%}

    {%- set target_relation = this.incorporate(type='table') -%}

    {{ run_hooks(pre_hooks) }}

    -- Drop and recreate the external table definition
    {% call statement('main') %}
        {% if load_cached_relation(this) is not none %}
            drop table if exists {{ external_schema }}.{{ this.identifier }};
        {% endif %}

        create external table {{ external_schema }}.{{ this.identifier }}
        ({{ sql }})
        {% if partition_cols %}
        partitioned by ({{ partition_cols | join(', ') }})
        {% endif %}
        stored as {{ file_format }}
        location '{{ s3_location }}';

        {% if partition_cols %}
        -- Auto-discover partitions
        msck repair table {{ external_schema }}.{{ this.identifier }};
        {% endif %}
    {% endcall %}

    {{ run_hooks(post_hooks) }}

    {{ return({'relations': [target_relation]}) }}

{% endmaterialization %}

Macro Testing and Documentation

Documenting Macros

yaml

# macros/schema.yml
macros:
  - name: pivot
    description: >
      Generates a pivot table using CASE/SUM expressions. Redshift-compatible.
      Use instead of native PIVOT which Redshift does not support.
    arguments:
      - name: column
        type: string
        description: "The column to pivot on"
      - name: values
        type: list
        description: "Distinct values to pivot into columns"
      - name: agg
        type: string
        description: "Aggregation function (default: sum)"
      - name: alias
        type: boolean
        description: "Whether to alias generated columns (default: true)"

  - name: unload_to_s3
    description: "UNLOAD a relation to S3 in Parquet or CSV format"
    arguments:
      - name: relation
        type: relation
        description: "The dbt relation to unload"
      - name: s3_path
        type: string
        description: "S3 destination path (must end with /)"
      - name: iam_role
        type: string
        description: "IAM role ARN with S3 write and Redshift UNLOAD permissions"

Testing Macros via `dbt run-operation`

bash

# Test the date_spine macro output
dbt run-operation date_spine --args "{'start_date': '2024-01-01', 'end_date': '2024-01-07'}"

# Test UNLOAD macro in dry-run mode (check SQL compilation)
dbt compile --select fct_orders  # ensure model compiles first

5 Practice Questions

Practice Question

Why is `namespace()` needed when updating a variable inside a Jinja2 `{% for %}` loop?

Practice Question

In a custom materialization, what does `config.require('my_param')` do differently from `config.get('my_param')`?

Practice Question

The `{% materialization my_mat, adapter='redshift' %}` signature means what?

Practice Question

You want to UNLOAD a Redshift table to S3 as Parquet, partitioned by `order_date`. Which dbt mechanism is most appropriate?

Practice Question

What guard is essential around `run_query()` calls in macros to prevent them from executing during dbt parse or compile phases?

Practice Question

Where should custom materialization files be stored in a dbt project?

✅Success

Key Takeaways

Use namespace() for mutable state inside Jinja2 {% for %} loops — plain variable assignment does not escape loop scope.
Always guard run_query() calls with {% if execute %} to prevent execution during parse/compile phases.
config.require() raises a compilation error for missing configs; config.get() returns None or a default — use require() for mandatory materialization parameters.
The adapter='redshift' signature in {% materialization %} creates a Redshift-specific implementation; other adapters fall back to the default implementation.
UNLOAD to S3 is best implemented as a post-hook macro, keeping the transformation logic separate from the export step.
Document macros in macros/schema.yml — they appear in dbt docs and can be tested via dbt run-operation.

Progreso55%

Model Contracts, Unit Tests, and Data Quality

Snapshots, Seeds, and Advanced Sources

Contenido del Curso

1. dbt-core Architecture and the dbt-redshift Adapter 2. Redshift Performance Patterns: Sort Keys, Dist Styles, and Compression 3. Materializations Deep Dive: Tables, Views, Late-Binding Views, and Redshift Materialized Views 4. Advanced Incremental Models and the Microbatch Strategy 5. Model Contracts, Unit Tests, and Data Quality 6. Advanced Jinja2 Macros and Custom Materializations 7. Snapshots, Seeds, and Advanced Sources 8. CI/CD Pipelines and Slim CI on Redshift 9. dbt Docs: Documentation, Lineage, and the dbt Docs v2 Catalog 10. Essential dbt Packages for Production Projects 11. dbt Fusion Engine and dbt Wizard

Advanced Jinja2 Macros and Custom Materializations

Advanced Jinja2 Macros and Custom Materializations

Jinja2 in dbt: What's Available

Macro Fundamentals: Beyond the Basics

Variable Scoping with namespace()

Returning Values from Macros

Production Macro Library for Redshift

1. Dynamic Union of Multiple Sources

2. Pivot Macro (Redshift-safe)

3. Generate Date Spine

4. Redshift UNLOAD to S3 Macro

5. Adaptive Sort Key Recommendation Macro

Custom Materializations

Anatomy of a Custom Materialization

Custom Materialization: Redshift External Table Refresh

Macro Testing and Documentation

Documenting Macros

Testing Macros via dbt run-operation

5 Practice Questions

Key Takeaways

Contenido del Curso

Testing Macros via `dbt run-operation`