Analytics Engineer

Analytics Engineer

The agent operates as a senior analytics engineer, building scalable dbt transformation layers, designing dimensional models, writing tested SQL, and managing semantic-layer metric definitions.

Clarify First

Before building the models, confirm these inputs. If any is unknown or vague, ASK — do not assume:

• [ ] Required grain + downstream consumers — the row grain of the target model and who queries it (dashboard, notebook, reverse-ETL) (drives the dimensional model and materialization)
• [ ] Source tables and freshness — which sources exist, their keys, and load cadence (determines staging models and incremental logic)
• [ ] Data volume + refresh SLA — table size and how often it must rebuild (selects view vs. table vs. incremental materialization)

Stop rule: ask only the 2-3 that most change the output. If the user says "just draft it," proceed and list your assumptions at the top of the artifact.

Workflow

1. Understand the data request -- Identify the business question, required grain, and downstream consumers (dashboard, notebook, reverse-ETL). Confirm source tables exist and check freshness. 2. Design the dimensional model -- Choose star or snowflake schema. Map source entities to dimension and fact tables at the correct grain. Document grain, primary keys, and foreign keys. 3. Build staging models -- One stg_ model per source table. Rename columns, cast types, filter soft-deletes, and add metadata columns. Validate: dbt build --select stg_*. 4. Build intermediate models -- Encapsulate reusable business logic in int_ models (e.g., int_orders_enriched). Keep each CTE single-purpose. 5. Build mart models -- Create dim_ and fct_ models for consumption. Configure materialization (view for staging, incremental for large facts, table for small marts). 6. Add tests and documentation -- Every primary key gets unique + not_null. Foreign keys get relationships. Add accepted_values for enums. Write model descriptions in YAML. 7. Define semantic-layer metrics -- Register metrics (sum, average, count_distinct) with time grains and dimension slices so BI consumers get a single source of truth. 8. Validate end-to-end -- Run dbt build, confirm test pass rate = 100%, check row counts against source, and verify dashboard numbers match.

dbt Project Structure

analytics/
  dbt_project.yml
  models/
    staging/          # stg_<source>__<table>.sql  (one per source table)
    intermediate/     # int_<entity>_<verb>.sql     (reusable logic)
    marts/
      core/           # dim_*.sql, fct_*.sql        (consumption-ready)
      marketing/
      finance/
  macros/             # Reusable Jinja helpers
  tests/              # Custom generic + singular tests
  seeds/              # Static CSV lookups
  snapshots/          # SCD Type 2 captures

Concrete Example: Customer Dimension

Staging model (models/staging/crm/stg_crm__customers.sql):

WITH source AS (
    SELECT * FROM {{ source('crm', 'customers') }}
),

renamed AS (
    SELECT
        id                          AS customer_id,
        TRIM(LOWER(name))           AS customer_name,
        TRIM(LOWER(email))          AS email,
        created_at::timestamp       AS created_at,
        updated_at::timestamp       AS updated_at,
        is_active::boolean          AS is_active,
        _fivetran_synced            AS _loaded_at
    FROM source
    WHERE _fivetran_deleted = false
)

SELECT * FROM renamed

Mart model (models/marts/core/dim_customer.sql):

WITH customers AS (
    SELECT * FROM {{ ref('stg_crm__customers') }}
),

customer_orders AS (
    SELECT
        customer_id,
        MIN(order_date)  AS first_order_date,
        MAX(order_date)  AS most_recent_order_date,
        COUNT(*)         AS lifetime_orders,
        SUM(order_amount) AS lifetime_value
    FROM {{ ref('stg_orders__orders') }}
    GROUP BY customer_id
),

final AS (
    SELECT
        c.customer_id,
        c.customer_name,
        c.email,
        c.created_at,
        co.first_order_date,
        co.most_recent_order_date,
        co.lifetime_orders,
        co.lifetime_value,
        CASE
            WHEN co.lifetime_value >= 10000 THEN 'platinum'
            WHEN co.lifetime_value >= 5000  THEN 'gold'
            WHEN co.lifetime_value >= 1000  THEN 'silver'
            ELSE 'bronze'
        END AS customer_tier
    FROM customers c
    LEFT JOIN customer_orders co
        ON c.customer_id = co.customer_id
)

SELECT * FROM final

Test configuration (models/marts/core/_core__models.yml):

version: 2
models:
  - name: dim_customer
    description: Customer dimension with lifetime order metrics and tier classification.
    columns:
      - name: customer_id
        tests: [unique, not_null]
      - name: email
        tests: [unique, not_null]
      - name: customer_tier
        tests:
          - accepted_values:
              values: ['platinum', 'gold', 'silver', 'bronze']
      - name: lifetime_value
        tests:
          - dbt_utils.expression_is_true:
              expression: ">= 0"

Incremental Fact Table Pattern

-- models/marts/core/fct_orders.sql
{{
    config(
        materialized='incremental',
        unique_key='order_id',
        partition_by={'field': 'order_date', 'data_type': 'date'},
        cluster_by=['customer_id', 'product_id']
    )
}}

WITH orders AS (
    SELECT * FROM {{ ref('stg_orders__orders') }}
    {% if is_incremental() %}
    WHERE order_date >= (SELECT MAX(order_date) FROM {{ this }})
    {% endif %}
),

order_items AS (
    SELECT * FROM {{ ref('stg_orders__order_items') }}
),

final AS (
    SELECT
        o.order_id,
        o.order_date,
        o.customer_id,
        oi.product_id,
        o.store_id,
        oi.quantity,
        oi.unit_price,
        oi.quantity * oi.unit_price AS line_total,
        o.discount_amount,
        o.tax_amount,
        o.total_amount
    FROM orders o
    INNER JOIN order_items oi ON o.order_id = oi.order_id
)

SELECT * FROM final

Materialization Strategy

Semantic-Layer Metric Definition

# models/marts/core/_core__metrics.yml
metrics:
  - name: revenue
    label: Total Revenue
    model: ref('fct_orders')
    calculation_method: sum
    expression: total_amount
    timestamp: order_date
    time_grains: [day, week, month, quarter, year]
    dimensions: [customer_tier, product_category, store_region]
    filters:
      - field: is_cancelled
        operator: '='
        value: 'false'

  - name: average_order_value
    label: Average Order Value
    model: ref('fct_orders')
    calculation_method: average
    expression: total_amount
    timestamp: order_date
    time_grains: [day, week, month]

Useful Macros

-- macros/cents_to_dollars.sql
{% macro cents_to_dollars(column_name) %}
    ({{ column_name }} / 100.0)::decimal(18,2)
{% endmacro %}

-- macros/get_incremental_filter.sql
{% macro get_incremental_filter(column_name, lookback_days=3) %}
    {% if is_incremental() %}
        WHERE {{ column_name }} >= (
            SELECT DATEADD(day, -{{ lookback_days }}, MAX({{ column_name }}))
            FROM {{ this }}
        )
    {% endif %}
{% endmacro %}

CI/CD: Slim CI for Pull Requests

# Only run modified models and their downstream dependents
dbt run  --select state:modified+ --defer --state ./target-base
dbt test --select state:modified+ --defer --state ./target-base

For full CI/CD pipeline configuration, see REFERENCE.md.

Reference Materials

• REFERENCE.md -- Extended patterns: source config, custom tests, CI/CD workflows, exposures, documentation templates
• references/modeling_patterns.md -- Data modeling best practices
• references/dbt_style_guide.md -- SQL and dbt conventions
• references/testing_guide.md -- Testing strategies
• references/optimization.md -- Performance tuning

Scripts

python scripts/impact_analyzer.py --model dim_customer
python scripts/schema_diff.py --source prod --target dev
python scripts/doc_generator.py --format markdown
python scripts/quality_scorer.py --model fct_orders

Tool Reference

Troubleshooting

Success Criteria

• All dbt models pass dbt build with a 100% test pass rate before merging to production.
• Every model has a YAML description and at least one test per primary key (unique + not_null).
• Incremental models process new data in under 5 minutes for tables up to 100M rows.
• Schema drift between prod and dev environments is detected and reviewed before each release.
• quality_scorer.py reports >= 80/100 for every mart model.
• Downstream dashboards refresh within SLA (< 5 s load time) after transformation runs complete.
• Semantic-layer metrics are the single source of truth -- no ad-hoc metric calculations exist in BI tools.

Scope & Limitations

In scope: dbt project design, dimensional modeling (Kimball methodology), SQL transformation logic, data testing, semantic-layer metric definition, CI/CD for dbt, and warehouse query optimization.

Out of scope: Raw data ingestion and extraction (ELT/ETL orchestration tools like Fivetran or Airbyte), data infrastructure provisioning, BI tool configuration beyond semantic-layer integration, and real-time streaming pipelines.

Limitations: The Python tools operate on dbt manifest/catalog JSON artifacts and do not query the warehouse directly. Scoring heuristics in quality_scorer.py use rule-based deductions that may not cover every project convention. All scripts use the Python standard library only -- no external dependencies required.

Integration Points

• Data Engineer (engineering/senior-data-engineer): Coordinates on source table contracts, ingestion SLAs, and schema change notifications.
• Business Intelligence (data-analytics/business-intelligence): Consumes mart models and semantic-layer metrics; dashboard specs reference model outputs.
• Data Analyst (data-analytics/data-analyst): Writes ad-hoc queries against mart models; reports data quality issues back to the analytics engineer.
• MLOps Engineer (data-analytics/ml-ops-engineer): Feature engineering pipelines may depend on intermediate or mart models as upstream inputs.
• CI/CD Workflows (templates/): Slim CI patterns (state:modified+) integrate into GitHub Actions or similar runners for automated PR validation.

Analytics Engineer -- Extended Reference

Source Configuration

# models/staging/crm/_crm__sources.yml
version: 2
sources:
  - name: crm
    description: Customer relationship management system
    database: raw
    schema: crm
    loader: fivetran
    loaded_at_field: _fivetran_synced
    freshness:
      warn_after: {count: 12, period: hour}
      error_after: {count: 24, period: hour}
    tables:
      - name: customers
        description: Customer master data
        columns:
          - name: id
            description: Primary key
            tests: [unique, not_null]
          - name: email
            tests: [unique]

Custom Generic Tests

-- tests/assert_positive_amount.sql
{% test positive_amount(model, column_name) %}
SELECT {{ column_name }}
FROM {{ model }}
WHERE {{ column_name }} < 0
{% endtest %}

-- tests/generic/assert_row_count_equal.sql
{% test row_count_equal(model, compare_model) %}
WITH source_count AS (
    SELECT COUNT(*) AS cnt FROM {{ model }}
),
compare_count AS (
    SELECT COUNT(*) AS cnt FROM {{ ref(compare_model) }}
)
SELECT *
FROM source_count
CROSS JOIN compare_count
WHERE source_count.cnt != compare_count.cnt
{% endtest %}

Additional Macros

-- macros/generate_schema_name.sql
{% macro generate_schema_name(custom_schema_name, node) -%}
    {%- set default_schema = target.schema -%}
    {%- if custom_schema_name is none -%}
        {{ default_schema }}
    {%- else -%}
        {{ default_schema }}_{{ custom_schema_name | trim }}
    {%- endif -%}
{%- endmacro %}

-- macros/pivot_values.sql
{% macro pivot_values(column_name, values, alias_prefix='') %}
    {% for value in values %}
        SUM(CASE WHEN {{ column_name }} = '{{ value }}' THEN 1 ELSE 0 END)
            AS {{ alias_prefix }}{{ value | lower | replace(' ', '_') }}
        {% if not loop.last %},{% endif %}
    {% endfor %}
{% endmacro %}

Exposures

# models/exposures.yml
version: 2
exposures:
  - name: executive_dashboard
    type: dashboard
    maturity: high
    url: https://tableau.company.com/views/executive
    description: Executive KPI dashboard
    depends_on:
      - ref('fct_orders')
      - ref('dim_customer')
      - ref('dim_product')
    owner:
      name: Analytics Team
      email: analytics@company.com

GitHub Actions CI/CD Pipeline

# .github/workflows/dbt.yml
name: dbt CI/CD
on:
  pull_request:
    branches: [main]
  push:
    branches: [main]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - uses: actions/setup-python@v4
        with:
          python-version: '3.10'
      - run: pip install dbt-snowflake
      - run: dbt deps
      - run: dbt compile --target ci
      - run: dbt test --target ci

  deploy:
    needs: test
    if: github.ref == 'refs/heads/main'
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - run: dbt run --target prod
      - run: dbt test --target prod

Query Optimization: Pre-aggregate Pattern

-- Before: expensive window function on full table
SELECT order_id, customer_id, order_date,
    SUM(amount) OVER (PARTITION BY customer_id ORDER BY order_date) AS running_total
FROM orders;

-- After: pre-aggregate then join
WITH daily_totals AS (
    SELECT customer_id, order_date, SUM(amount) AS daily_amount
    FROM orders
    GROUP BY customer_id, order_date
),
running_totals AS (
    SELECT customer_id, order_date,
        SUM(daily_amount) OVER (PARTITION BY customer_id ORDER BY order_date) AS running_total
    FROM daily_totals
)
SELECT o.order_id, o.customer_id, o.order_date, rt.running_total
FROM orders o
JOIN running_totals rt
    ON o.customer_id = rt.customer_id AND o.order_date = rt.order_date;

Model Documentation Template

models:
  - name: fct_orders
    description: |
      Order fact table containing one row per order line item.

      ## Business Logic
      - Orders with status 'cancelled' are excluded
      - Amounts are in USD
      - Tax is calculated at time of order

      ## Usage

SELECT * FROM {{ ref('fct_orders') }} WHERE order_date >= '2024-01-01'

      ## Dependencies
      - stg_orders__orders
      - stg_orders__order_items

dbt Style Guide & Conventions

Consistent project structure and SQL style are critical for maintaining large-scale dbt architectures.

1. Project Layering

The dbt project must follow a strict three-layer architecture:

Staging (models/staging/)

• Forms a strict 1:1 relationship with source tables.
• Purpose: Renaming columns, casting data types, standardizing booleans, and applying trim/lower logic.
• Constraints:
• Strictly NO JOINs or aggregations allowed!
• Must use the source('schema', 'table') macro.
• Materialization: view
• Naming Prefix: stg_<source_name>__<table_name>.sql

Intermediate (models/intermediate/)

• Encapsulates complex transformational logic.
• Purpose: Consolidating multi-source data, joining lookup tables, and executing window functions.
• Materialization: ephemeral (compiled inside downstream CTEs) or view if referenced multiple times.
• Naming Prefix: int_<entity_name>_<verb>.sql (e.g., int_users_enriched.sql)

Marts (models/marts/)

• The business-facing dimensional and fact models.
• Purpose: Serving clean, optimized data to BI tools.
• Materialization: table (or incremental for massive fact tables).
• Naming Prefix: dim_<entity>.sql or fct_<event>.sql

---

2. SQL Formatting and Syntax

No Implicit Columns

• NEVER use SELECT * against source or ref tables in Production models (outside of simple staging passthroughs).
• The terminal SELECT * FROM final CTE pattern is acceptable as it references an explicitly projected CTE.
• Why: Implicit lists break dbt data contracts during schema drift and inflate network/memory payloads. Always project columns explicitly in CTEs.

CTE Structure

• All models must use Common Table Expressions (CTEs), ending with a final SELECT * FROM final block.
• Why: Separating logic into modular CTE blocks allows for isolation and step-by-step debugging.

Example Standard Model:

WITH customers AS (
    SELECT * FROM {{ ref('stg_salesforce__customers') }}
),
orders AS (
    SELECT * FROM {{ ref('stg_stripe__orders') }}
),
customer_orders AS (
    SELECT
        customer_id,
        MIN(order_date) AS first_order_date,
        SUM(amount) AS lifetime_value
    FROM orders
    GROUP BY customer_id
),
final AS (
    SELECT
        c.customer_id,
        c.customer_name,
        co.first_order_date,
        co.lifetime_value
    FROM customers AS c
    LEFT JOIN customer_orders AS co
        ON c.customer_id = co.customer_id
)

SELECT * FROM final;

Jinja and Macros

• Store reusable logic (e.g., converting cents to dollars, generating schema names) inside the macros/ folder.
• Use target.name conditionals to limit data queried in dev environments to save operational costs.

Data Modeling Patterns

This guide formalizes the dimensional data modeling conventions expected within the analytics engineering workflow.

1. Dimensional Architectures

The Star Schema

The Star Schema is the universally preferred architecture for analytics. It relies on a single Fact Table surrounded by highly denormalized Dimension Tables.

• Performance: Minimizes complex multi-table joins.
• Understandability: Business users can natively grok the "nouns" (dimensions) and "verbs" (facts) of the business.
• Rule: Never use a Snowflake schema unless explicitly required by a BI tool limitation. A dimension table should contain all attributes required to slice the fact table, regardless of normalization redundancy.

One Big Table (OBT)

• Warning: Do not use OBT approaches. While columnar databases handle them well, they force disparate business processes into monolithic structures, resulting in extreme matrix sparsity and complex workarounds.

---

2. Dimension Tables

Dimensions (the "who, what, where, when") provide the text attributes used to filter and group fact data.

Surrogate Keys

• Rule: Every dimension table MUST have a meaningless, auto-incrementing integer (or hashed integer) as its primary key.
• Why: Protects the warehouse from upstream source system changes (e.g., if a natural key is recycled) and enables Slowly Changing Dimension (SCD) capabilities.
• Implementation: Use dbt_utils.generate_surrogate_key(['natural_key']).

Hierarchies

• Do not normalize hierarchies. The Dim_Product table should contain Category, Sub_Category, Brand, and SKU directly on the single row.

Slowly Changing Dimensions (SCDs)

• SCD Type 1 (Overwrite): Use when history is irrelevant. Edits overwrite the old data.
• SCD Type 2 (Versioning): Required for compliance and historical accuracy. Create a new row when an attribute changes. Requires is_active boolean, effective_date, and expiration_date columns.
• For implementing SCD Type 2 in dbt, utilize the `snapshots` feature rather than manual logic.

---

3. Fact Tables

Fact tables capture the measurable, quantitative metrics of a business event.

Target Atomic Grain

• Define the lowest possible granularity. The fact table should represent the most atomic level of an event (e.g., one row per order line item, not one row per order).

Additive Facts Only

• Rule: Store fully additive measures (e.g., dollars_sold, quantity_ordered).
• Rule: Non-additive measures (e.g., gross_margin_percentage, conversion_rate) must NOT be stored in the database. Instead, store the additive components (revenue, cost) and perform the ratio division dynamically in the BI tool.

Degenerate Dimensions

• Rule: For extremely high-cardinality values with no descriptive attributes (e.g., transaction_id, invoice_number), store them directly on the fact table. Do not create a dimension table just to hold an ID.

No NULL Foreign Keys

• Rule: A fact table must never contain a NULL foreign key. Inner joins drop records with NULL keys, silently obliterating revenue metrics.
• Fix: Direct NULL foreign keys to an established placeholder dimension row (e.g., ID -1 mapped to "Unknown" or "N/A").

SQL Query Optimization Patterns

This guide outlines antipatterns that cripple data warehouse performance and details the strategies required to remediate them.

1. Eliminate Spaghetti Queries

• The Antipattern: Writing a monolithic, multi-thousand-line SELECT statement attempting to aggregate, filter, and join a dozen tables at once to fulfill an entire reporting requirement.
• The Issue: Massive queries confuse the warehouse optimizer. The engine often allocates incorrect memory logic, resulting in unintentional Cartesian products (cross-joins) that spin compute resources into infinity.
• The Fix: Adopt a Divide and Conquer methodology. Break complex operations down into distinct sequential dbt Intermediate models (int_ layer) or distinct, linearly progressive Common Table Expressions (CTEs) representing granular steps.

2. Eliminate EAV and Jaywalking Joins

• Jaywalking: Storing comma-separated delimited IDs inside a single column and using JOIN ON a LIKE '%' || b || '%' to connect them.
• EAV (Entity-Attribute-Value): Storing schema-less data in three vertical columns (entity_id, attribute_name, value).
• The Issue: Both patterns completely bypass the relational database indexing structures, forcing horizontal table scans that are mathematically exponential.
• The Fix: Abstract many-to-many elements into an explicit Intersection Table natively supporting primary-to-foreign key bridging.

3. Ambiguous Aggregations

• The Antipattern: Selecting raw columns alongside aggregate measures without explicitly grouping them, or adding extraneous descriptive fields to a GROUP BY clause just to pass syntax.
• The Issue: It results in pseudo-random value selection for the non-grouped rows or inadvertently alters the grain of the calculation, resulting in duplicated or dropped financial measures.
• The Fix: Follow the single-value rule. Only group on the absolute core dimension keys. For descriptive descriptors, wrap them in deterministic functions like MAX(), or calculate the core aggregates in a nested CTE and JOIN the descriptors afterward.

4. Unoptimized Randomization

• The Antipattern: SELECT * FROM massive_table ORDER BY RAND() LIMIT 10
• The Issue: The database must assign a random floating-point integer to every single row in the millions-row facts table, execute a global sort on the entire table in memory, select the first 10, and discard the rest.
• The Fix: Avoid RAND(). Instead use windowing sequencing operations:

-- More optimal approach for sampling:
WITH CTE AS (
  SELECT *, ROW_NUMBER() OVER(ORDER BY event_timestamp) AS rn
  FROM massive_table
)
SELECT * FROM CTE WHERE rn % 100 = 0; -- takes 1% roughly

5. Incremental Processing

For facts exceeding ten million records, daily rebuilds consume massive resources. Configure fact models as materialized='incremental' inside the configuration block.

• Read only the last 1-3 days of data utilizing the {% if is_incremental() %} Jinja macro condition.
• Update matching records utilizing unique_key.

Testing & CI/CD Guide

Ensuring data pipeline integrity requires rigorous coverage at all execution vectors.

1. Generic Tests

Generic tests are globally scalable validations declared within .yml schema files.

• Unique: Apply to all primary keys to ensure the grain of the model is respected.
• Not_null: Apply to all primary keys and critical business fields (e.g. amount).
• Accepted_values: Use to validate enum-type fields (e.g., ensuring order_status can only equal pending, shipped, cancelled).
• Relationships: Apply to all foreign keys to validate referential integrity against parent dimension tables.

Implementation Example:

models:
  - name: fct_orders
    columns:
      - name: order_id
        tests:
          - unique
          - not_null
      - name: customer_id
        tests:
          - relationships:
              to: ref('dim_customers')
              field: customer_id
      - name: status
        tests:
          - accepted_values:
              values: ['placed', 'shipped', 'completed']

2. Testing Sources First

Do not wait until the transformation is complete to discover bad data. Apply tests directly to the raw sources inside the _sources.yml configurations. Isolating upstream drops reduces debug turnaround.

3. Singular Tests

When business logic requires complex checks that generic .yml blocks can't accommodate, write explicit Singular Tests.

• Store these as .sql files within the tests/ directory.
• The test file should SELECT the failing condition. If the query returns 0 rows, the test passes.

Example: tests/assert_positive_amount.sql

-- Ensure that the revenue amount is never negative
SELECT
    order_id,
    revenue_amount
FROM {{ ref('fct_orders') }}
WHERE revenue_amount < 0

4. Continuous Integration (CI/CD)

Pull Request Workflows

Never merge directly into main. The Git workflow should mandate PRs, and the CI environment must automatically run building and testing.

Slim CI

Running a full dbt build against thousands of models on every PR commits is too slow and expensive.

• Rule: Utilize Slim CI workflows comparing state against production.
• Command: dbt build --select state:modified+
• Result: Instructs the environment to only build and test models whose code changed in the PR branch, along with any downstream models that depend on them.