advanced⏱70 minLição 6 de 11

Macros Avançadas em Jinja2 e Materializações Customizadas

Escreva macros Jinja2 de nível profissional para padrões SQL DRY, construa bibliotecas de macros reutilizáveis e crie materializações customizadas para padrões específicos do Redshift como logs de eventos append-only e refresh de tabelas externas.

Macros Avançadas em Jinja2 e Materializações Customizadas

Jinja2 é o motor de templates do dbt — ele transforma SQL de uma linguagem de consulta estática em uma linguagem dinâmica e composável. Dominar Jinja2 é a diferença entre um projeto que escala para 500 modelos com uma equipe pequena e um que se torna insustentável. Este módulo cobre padrões de macro de nível profissional e materializações customizadas para casos de uso específicos do Redshift.

O Jinja2 permite que você escreva SQL parametrizado, com loops, condicionais e reutilização de código. Sem ele, você teria que copiar e colar SQL repetitivo centenas de vezes, tornando a manutenção um pesadelo.

Jinja2 no dbt: O que Está Disponível

O dbt estende o Jinja2 com suas próprias funções e objetos. As três camadas são:

┌─────────────────────────────────────┐
│  Funções específicas do dbt          │
│  ref(), source(), config(), this     │
│  run_query(), log(), execute        │
├─────────────────────────────────────┤
│  Jinja2 padrão                       │
│  {% if %}, {% for %}, {% set %}      │
│  filters, tests, controle de whitespace │
├─────────────────────────────────────┤
│  Built-ins Python expostos pelo Jinja2│
│  range(), namespace(), loop.*        │
└─────────────────────────────────────┘

Fundamentos de Macro: Além do Básico

Escopo de Variáveis com namespace()

Variáveis de loop do Jinja2 não atualizam o escopo externo sem namespace():

sql

-- macros/utils/string_utils.sql

-- ERRADO: result é sempre string vazia após o loop
{% macro join_columns_wrong(columns) %}
    {% set result = '' %}
    {% for col in columns %}
        {% set result = result ~ col %}  {# Isto NÃO atualiza result externo #}
    {% endfor %}
    {{ result }}
{% endmacro %}

-- CORRETO: use namespace() para estado mutável dentro de 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 %}

Retornando Valores de 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 %}

Usando o valor retornado:

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

Biblioteca de Macros de Produção para Redshift

1. Union Dinâmico de Múltiplas Fontes

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 %}

Uso:

sql

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

2. Macro Pivot (Seguro para Redshift)

O Redshift não tem um PIVOT nativo. Esta macro gera o padrão CASE/SUM:

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. Gerar 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. Macro UNLOAD do Redshift para S3

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 %}

Uso como operação:

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. Macro de Recomendação de Sort Keys Adaptativa

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 %}

Materializações Customizadas

Materializações customizadas permitem definir novos padrões de persistência além dos integrados do dbt. São blocos Jinja2 armazenados em macros/materializations/.

Anatomia de uma Materialização Customizada

sql

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

    {# Configuração obrigatória #}
    {%- set target_relation = this.incorporate(type='table') -%}

    {# Obter o adapter de execução #}
    {%- set existing_relation = load_cached_relation(this) -%}

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

    {# Criar tabela na primeira execução #}
    {% if existing_relation is none %}
        {% call statement('main') %}
            {{ create_table_as(false, target_relation, sql) }}
        {% endcall %}

    {# Append nas execuções subsequentes #}
    {% else %}
        {% call statement('main') %}
            insert into {{ target_relation }}
            ({{ sql }})
        {% endcall %}
    {% endif %}

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

    {# Atualizar o cache de relations #}
    {{ return({'relations': [target_relation]}) }}

{% endmaterialization %}

Usando a materialização customizada:

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

Materialização Customizada: Refresh de Tabela Externa Redshift Spectrum

Para tabelas externas do Redshift Spectrum que precisam de refresh periódico de metadados:

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 e recria a definição da tabela externa
    {% 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-descobrir partições
        msck repair table {{ external_schema }}.{{ this.identifier }};
        {% endif %}
    {% endcall %}

    {{ run_hooks(post_hooks) }}

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

{% endmaterialization %}

Teste e Documentação de Macros

Documentando Macros

yaml

# macros/schema.yml
macros:
  - name: pivot
    description: >
      Gera uma tabela pivot usando expressões CASE/SUM. Compatível com Redshift.
      Use em vez do PIVOT nativo que o Redshift não suporta.
    arguments:
      - name: column
        type: string
        description: "A coluna para pivotear"
      - name: values
        type: list
        description: "Valores distintos para transformar em colunas"
      - name: agg
        type: string
        description: "Função de agregação (padrão: sum)"
      - name: alias
        type: boolean
        description: "Se deve aliasear colunas geradas (padrão: true)"

  - name: unload_to_s3
    description: "UNLOAD de uma relation para S3 em formato Parquet ou CSV"
    arguments:
      - name: relation
        type: relation
        description: "A relação dbt para exportar"
      - name: s3_path
        type: string
        description: "Caminho de destino S3 (deve terminar com /)"
      - name: iam_role
        type: string
        description: "ARN da IAM Role com permissões de escrita S3 e UNLOAD Redshift"

Testando Macros via `dbt run-operation`

bash

# Testar a saída da macro date_spine
dbt run-operation date_spine --args "{'start_date': '2024-01-01', 'end_date': '2024-01-07'}"

# Testar macro UNLOAD em modo dry-run (verificar compilação SQL)
dbt compile --select fct_orders  # garantir que o modelo compila primeiro

6 Perguntas de Prática

Practice Question

Por que `namespace()` é necessário ao atualizar uma variável dentro de um loop `{% for %}` no Jinja2?

Practice Question

Em uma materialização customizada, o que `config.require('my_param')` faz de diferente de `config.get('my_param')`?

Practice Question

A assinatura `{% materialization my_mat, adapter='redshift' %}` significa o quê?

Practice Question

Você quer exportar (UNLOAD) uma tabela Redshift para S3 como Parquet, particionada por `order_date`. Qual mecanismo dbt é mais apropriado?

Practice Question

Qual proteção é essencial em torno de chamadas `run_query()` em macros para evitar execução durante as fases de parse ou compilação do dbt?

Practice Question

Onde os arquivos de materialização customizada devem ser armazenados em um projeto dbt?

✅Success

Principais Conclusões

Use namespace() para estado mutável dentro de loops {% for %} no Jinja2 — atribuição simples de variável não escapa do escopo do loop.
Sempre proteja chamadas run_query() com {% if execute %} para prevenir execução durante fases de parse/compilação.
config.require() levanta erro de compilação para configs ausentes; config.get() retorna None ou um padrão — use require() para parâmetros obrigatórios de materialização.
A assinatura adapter='redshift' em {% materialization %} cria uma implementação específica para Redshift; outros adapters usam a implementação default.
UNLOAD para S3 é melhor implementado como uma macro post-hook, mantendo a lógica de transformação separada da etapa de exportação.
Documente macros em macros/schema.yml — elas aparecem em dbt docs e podem ser testadas via dbt run-operation.

Progresso55%

Contratos de Modelo, Testes Unitários e Qualidade de Dados

Snapshots, Seeds e Sources Avançadas

Contéudo do Curso

1. Arquitetura do dbt-core e o Adapter dbt-redshift 2. Padrões de Performance Redshift: Sort Keys, Dist Styles e Compressão 3. Materializações em Profundidade: Tabelas, Views, Late-Binding Views e Materialized Views do Redshift 4. Modelos Incrementais Avançados e a Estratégia Microbatch 5. Contratos de Modelo, Testes Unitários e Qualidade de Dados 6. Macros Avançadas em Jinja2 e Materializações Customizadas 7. Snapshots, Seeds e Sources Avançadas 8. Pipelines CI/CD e Slim CI no Redshift 9. dbt Docs: Documentação, Linhagem e o Catálogo dbt Docs v2 10. Pacotes dbt Essenciais para Projetos de Produção 11. Motor dbt Fusion e dbt Wizard

Macros Avançadas em Jinja2 e Materializações Customizadas

Macros Avançadas em Jinja2 e Materializações Customizadas

Jinja2 no dbt: O que Está Disponível

Fundamentos de Macro: Além do Básico

Escopo de Variáveis com namespace()

Retornando Valores de Macros

Biblioteca de Macros de Produção para Redshift

1. Union Dinâmico de Múltiplas Fontes

2. Macro Pivot (Seguro para Redshift)

3. Gerar Date Spine

4. Macro UNLOAD do Redshift para S3

5. Macro de Recomendação de Sort Keys Adaptativa

Materializações Customizadas

Anatomia de uma Materialização Customizada

Materialização Customizada: Refresh de Tabela Externa Redshift Spectrum

Teste e Documentação de Macros

Documentando Macros

Testando Macros via dbt run-operation

6 Perguntas de Prática

Principais Conclusões

Contéudo do Curso

Testando Macros via `dbt run-operation`