Building Internal Data Platforms: The Modern Data Stack in Practice

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

# docker-compose.yml (simplified — use official Airbyte install for production)
version: "3.7"
services:
  airbyte-server:
    image: airbyte/server:0.63.0
    environment:
      - DATABASE_URL=postgresql://airbyte:password@airbyte-db:5432/airbyte
    ports:
      - "8001:8001"
    depends_on:
      - airbyte-db

  airbyte-webapp:
    image: airbyte/webapp:0.63.0
    ports:
      - "8000:80"

  airbyte-db:
    image: postgres:13
    environment:
      POSTGRES_USER: airbyte
      POSTGRES_PASSWORD: password
      POSTGRES_DB: airbyte

1
2
3

# Official install
wget https://raw.githubusercontent.com/airbytehq/airbyte/master/run-ab-platform.sh
chmod +x run-ab-platform.sh && ./run-ab-platform.sh

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

# custom_source/source.py
from airbyte_cdk.sources import AbstractSource
from airbyte_cdk.sources.streams import Stream
from airbyte_cdk.models import SyncMode
import requests

class InternalAPIStream(Stream):
    primary_key = "id"

    def __init__(self, api_key: str, endpoint: str):
        self.api_key = api_key
        self.endpoint = endpoint

    @property
    def name(self):
        return "internal_api_records"

    def read_records(self, sync_mode, cursor_field=None, stream_slice=None,
                     stream_state=None):
        headers = {"Authorization": f"Bearer {self.api_key}"}
        page = 1
        while True:
            response = requests.get(
                f"{self.endpoint}?page={page}&limit=100",
                headers=headers
            )
            data = response.json()
            records = data.get("items", [])
            if not records:
                break
            yield from records
            page += 1

class SourceInternalAPI(AbstractSource):
    def check_connection(self, logger, config):
        try:
            response = requests.get(
                config["endpoint"] + "/health",
                headers={"Authorization": f"Bearer {config['api_key']}"}
            )
            return response.status_code == 200, None
        except Exception as e:
            return False, str(e)

    def streams(self, config):
        return [InternalAPIStream(config["api_key"], config["endpoint"])]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

# debezium-connector.json
{
  "name": "postgres-production-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres-prod",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "${secret:postgres-password}",
    "database.dbname": "production",
    "database.server.name": "production",
    "plugin.name": "pgoutput",
    "table.include.list": "public.orders,public.users,public.products",
    "slot.name": "debezium_slot",
    "publication.name": "debezium_pub",
    "transforms": "unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.unwrap.drop.tombstones": "false",
    "transforms.unwrap.delete.handling.mode": "rewrite",
    "topic.prefix": "cdc"
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

-- models/staging/stg_orders.sql
with source as (
    select * from {{ source('raw_postgres', 'orders') }}
),

renamed as (
    select
        id::varchar                          as order_id,
        user_id::varchar                     as user_id,
        status                               as order_status,
        total_amount_cents / 100.0           as total_amount_usd,
        created_at::timestamptz              as created_at,
        updated_at::timestamptz              as updated_at,
        -- metadata
        _airbyte_extracted_at                as _extracted_at

    from source
    where id is not null  -- exclude malformed records
)

select * from renamed

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

# models/staging/_staging_sources.yml
version: 2

sources:
  - name: raw_postgres
    database: raw
    schema: postgres_public
    loaded_at_field: _airbyte_extracted_at
    freshness:
      warn_after: {count: 6, period: hour}
      error_after: {count: 24, period: hour}
    tables:
      - name: orders
        description: "Raw orders table from production PostgreSQL"
        columns:
          - name: id
            description: "Primary key"
            tests:
              - not_null
              - unique
          - name: status
            tests:
              - accepted_values:
                  values: ['pending', 'processing', 'shipped', 'delivered', 'cancelled', 'refunded']

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

-- models/intermediate/int_orders_with_users.sql
with orders as (
    select * from {{ ref('stg_orders') }}
),

users as (
    select * from {{ ref('stg_users') }}
),

final as (
    select
        o.order_id,
        o.user_id,
        o.order_status,
        o.total_amount_usd,
        o.created_at,
        u.email              as user_email,
        u.country            as user_country,
        u.created_at         as user_signup_date,

        -- Derived fields
        datediff('day', u.created_at, o.created_at) as days_since_signup,
        case
            when datediff('day', u.created_at, o.created_at) <= 7
            then 'new_user'
            else 'returning_user'
        end as user_cohort

    from orders o
    left join users u using (user_id)
)

select * from final

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

-- models/marts/core/fct_orders.sql
{{
  config(
    materialized='incremental',
    unique_key='order_id',
    on_schema_change='append_new_columns'
  )
}}

with orders as (
    select * from {{ ref('int_orders_with_users') }}
    {% if is_incremental() %}
        where created_at > (select max(created_at) from {{ this }})
    {% endif %}
),

order_items as (
    select * from {{ ref('stg_order_items') }}
),

final as (
    select
        o.order_id,
        o.user_id,
        o.user_email,
        o.user_country,
        o.user_cohort,
        o.order_status,
        o.total_amount_usd,
        o.created_at,
        count(oi.item_id)           as item_count,
        sum(oi.quantity)            as total_quantity,
        -- revenue recognition (completed orders only)
        case
            when o.order_status = 'delivered'
            then o.total_amount_usd
            else 0
        end as recognized_revenue_usd

    from orders o
    left join order_items oi using (order_id)
    group by 1, 2, 3, 4, 5, 6, 7, 8
)

select * from final

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

# models/marts/core/_core_models.yml
version: 2

models:
  - name: fct_orders
    description: "One row per order with user and item details"
    tests:
      - dbt_utils.equal_rowcount:
          compare_model: ref('stg_orders')  # mart should have same count as source
    columns:
      - name: order_id
        tests:
          - not_null
          - unique

      - name: total_amount_usd
        description: "Order total in USD"
        tests:
          - not_null
          - dbt_utils.accepted_range:
              min_value: 0
              max_value: 50000

      - name: order_status
        tests:
          - accepted_values:
              values: ['pending', 'processing', 'shipped', 'delivered', 'cancelled', 'refunded']

      - name: user_id
        tests:
          - not_null
          - relationships:
              to: ref('dim_users')
              field: user_id

1
2
3
4
5
6
7

-- tests/assert_revenue_not_negative.sql
-- This test fails if any rows are returned
select
    order_id,
    recognized_revenue_usd
from {{ ref('fct_orders') }}
where recognized_revenue_usd < 0

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

-- macros/test_not_decreasing.sql
{% test not_decreasing(model, column_name, partition_by) %}
    with data as (
        select
            {{ partition_by }},
            {{ column_name }},
            lag({{ column_name }}) over (
                partition by {{ partition_by }}
                order by updated_at
            ) as prev_value
        from {{ model }}
    )
    select *
    from data
    where {{ column_name }} < prev_value
{% endtest %}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

# packages.yml
packages:
  - package: dbt-labs/dbt_utils
    version: [">=1.0.0", "<2.0.0"]
  - package: calogica/dbt_expectations
    version: [">=0.9.0", "<1.0.0"]
  - package: dbt-labs/audit_helper
    version: [">=0.9.0", "<1.0.0"]
  - package: dbt-labs/codegen
    version: [">=0.11.0", "<1.0.0"]

1

dbt deps  # install packages

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

- name: total_amount_usd
  tests:
    - dbt_expectations.expect_column_values_to_be_between:
        min_value: 0
        max_value: 50000
    - dbt_expectations.expect_column_mean_to_be_between:
        min_value: 40
        max_value: 200
    - dbt_expectations.expect_column_quantile_values_to_be_between:
        quantile: 0.95
        min_value: 100
        max_value: 1000

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

# helm values for airflow on k8s
# helm install airflow apache-airflow/airflow --values values.yaml
executor: KubernetesExecutor
dags:
  gitSync:
    enabled: true
    repo: https://github.com/your-org/data-platform-dags
    branch: main
    depth: 1
    wait: 60
    subPath: dags/

config:
  core:
    dags_are_paused_at_creation: "False"
    max_active_runs_per_dag: "1"
  scheduler:
    dag_dir_list_interval: "30"
  webserver:
    expose_config: "False"

workers:
  resources:
    limits:
      memory: "4Gi"
      cpu: "2"

postgresql:
  enabled: true

redis:
  enabled: true

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116

# dags/dbt_pipeline.py
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator, BranchPythonOperator
from airflow.providers.slack.operators.slack_webhook import SlackWebhookOperator
from airflow.utils.trigger_rule import TriggerRule
import subprocess
import json

DBT_PROJECT_DIR = "/opt/airflow/dbt"
DBT_PROFILES_DIR = "/opt/airflow/dbt"

default_args = {
    "owner": "data-platform",
    "depends_on_past": False,
    "retries": 2,
    "retry_delay": timedelta(minutes=5),
    "on_failure_callback": lambda ctx: notify_slack(ctx, "failure"),
}

with DAG(
    dag_id="dbt_production",
    default_args=default_args,
    description="Run dbt models and data quality checks",
    schedule="0 6 * * *",  # 6 AM UTC daily
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=["dbt", "data-quality"],
) as dag:

    # Source freshness check before running models
    check_freshness = BashOperator(
        task_id="check_source_freshness",
        bash_command=f"""
            cd {DBT_PROJECT_DIR}
            dbt source freshness \
                --profiles-dir {DBT_PROFILES_DIR} \
                --target prod
        """,
    )

    # Run staging models
    run_staging = BashOperator(
        task_id="run_staging_models",
        bash_command=f"""
            cd {DBT_PROJECT_DIR}
            dbt run \
                --select staging \
                --profiles-dir {DBT_PROFILES_DIR} \
                --target prod \
                --vars '{{"run_date": "{{{{ ds }}}}"}}'
        """,
    )

    # Test staging models
    test_staging = BashOperator(
        task_id="test_staging_models",
        bash_command=f"""
            cd {DBT_PROJECT_DIR}
            dbt test \
                --select staging \
                --profiles-dir {DBT_PROFILES_DIR} \
                --target prod
        """,
    )

    # Run intermediate and mart models
    run_marts = BashOperator(
        task_id="run_mart_models",
        bash_command=f"""
            cd {DBT_PROJECT_DIR}
            dbt run \
                --select intermediate+ \
                --profiles-dir {DBT_PROFILES_DIR} \
                --target prod
        """,
    )

    # Test mart models
    test_marts = BashOperator(
        task_id="test_mart_models",
        bash_command=f"""
            cd {DBT_PROJECT_DIR}
            dbt test \
                --select intermediate+ \
                --profiles-dir {DBT_PROFILES_DIR} \
                --target prod \
                --store-failures
        """,
    )

    # Generate and upload documentation
    generate_docs = BashOperator(
        task_id="generate_docs",
        bash_command=f"""
            cd {DBT_PROJECT_DIR}
            dbt docs generate \
                --profiles-dir {DBT_PROFILES_DIR} \
                --target prod
            aws s3 sync target/ s3://your-bucket/dbt-docs/ --delete
        """,
    )

    notify_success = SlackWebhookOperator(
        task_id="notify_success",
        http_conn_id="slack_webhook",
        message="""
:white_check_mark: dbt pipeline completed successfully
*Date*: {{ ds }}
*Duration*: {{ (execution_date - prev_execution_date).seconds // 60 }} minutes
        """,
        trigger_rule=TriggerRule.ALL_SUCCESS,
    )

    check_freshness >> run_staging >> test_staging >> run_marts >> test_marts >> generate_docs >> notify_success

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

# dags/ingestion_dags.py
from airflow import DAG
from airflow.providers.airbyte.operators.airbyte import AirbyteTriggerSyncOperator
from datetime import datetime, timedelta

# Configuration-driven DAG generation
CONNECTIONS = [
    {"name": "stripe", "connection_id": "abc-123", "schedule": "0 */6 * * *"},
    {"name": "salesforce", "connection_id": "def-456", "schedule": "0 8 * * *"},
    {"name": "hubspot", "connection_id": "ghi-789", "schedule": "0 8 * * *"},
    {"name": "postgres-prod", "connection_id": "jkl-012", "schedule": "*/30 * * * *"},
]

def create_ingestion_dag(config: dict) -> DAG:
    with DAG(
        dag_id=f"ingestion_{config['name']}",
        schedule=config["schedule"],
        start_date=datetime(2026, 1, 1),
        catchup=False,
        default_args={
            "retries": 3,
            "retry_delay": timedelta(minutes=2),
        },
        tags=["ingestion", config["name"]],
    ) as dag:

        sync = AirbyteTriggerSyncOperator(
            task_id=f"sync_{config['name']}",
            airbyte_conn_id="airbyte",
            connection_id=config["connection_id"],
            asynchronous=False,
            timeout=3600,
            wait_seconds=3,
        )

        return dag

# Register all DAGs
for config in CONNECTIONS:
    globals()[f"ingestion_{config['name']}"] = create_ingestion_dag(config)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

# dagster_pipeline.py
from dagster import asset, AssetIn, ScheduleDefinition, define_asset_job
from dagster_dbt import DbtCliResource, dbt_assets

import pandas as pd
from sqlalchemy import create_engine

@asset(
    description="Raw orders from production database",
    group_name="ingestion",
)
def raw_orders() -> pd.DataFrame:
    engine = create_engine("postgresql://prod-db/production")
    return pd.read_sql("SELECT * FROM orders WHERE updated_at > now() - interval '1 day'", engine)

@asset(
    ins={"raw_orders": AssetIn()},
    description="Cleaned and validated orders",
    group_name="staging",
)
def staged_orders(raw_orders: pd.DataFrame) -> pd.DataFrame:
    df = raw_orders.copy()
    df["total_amount_usd"] = df["total_amount_cents"] / 100
    df = df[df["id"].notna()]
    df = df.drop_duplicates(subset=["id"])
    return df

# dbt integration
dbt_resource = DbtCliResource(project_dir="/opt/dbt", profiles_dir="/opt/dbt")

@dbt_assets(manifest="/opt/dbt/target/manifest.json")
def dbt_project_assets(context, dbt: DbtCliResource):
    yield from dbt.cli(["build"], context=context).stream()

# Schedule
daily_job = define_asset_job("daily_pipeline", selection="*")
daily_schedule = ScheduleDefinition(
    job=daily_job,
    cron_schedule="0 6 * * *",
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85

# contracts/fct_orders.yaml
apiVersion: v1
kind: DataContract
metadata:
  name: fct_orders
  version: "2.1.0"
  owner: data-platform-team
  stakeholders:
    - analytics-team
    - finance-team
    - ml-team
  effective_date: "2026-01-15"
  review_date: "2026-07-15"

description: >
  Core orders fact table. One row per order.
  Source of truth for revenue, conversion, and order funnel metrics.

# Schema contract
schema:
  - name: order_id
    type: VARCHAR
    nullable: false
    unique: true
    description: "Surrogate key for the order"

  - name: user_id
    type: VARCHAR
    nullable: false
    description: "FK to dim_users.user_id"

  - name: total_amount_usd
    type: NUMERIC(12,2)
    nullable: false
    description: "Order total in USD at time of purchase"
    constraints:
      min: 0
      max: 50000

  - name: order_status
    type: VARCHAR
    nullable: false
    description: "Current order status"
    accepted_values:
      - pending
      - processing
      - shipped
      - delivered
      - cancelled
      - refunded

  - name: created_at
    type: TIMESTAMPTZ
    nullable: false
    description: "Order creation time in UTC"

  - name: recognized_revenue_usd
    type: NUMERIC(12,2)
    nullable: false
    description: "Revenue recognized per accounting rules. Non-zero only for delivered orders."

# Quality commitments
quality:
  freshness:
    warn_after: "6 hours"
    error_after: "24 hours"
  completeness:
    - column: order_id
      min_fill_rate: 1.0
    - column: user_id
      min_fill_rate: 0.99
  accuracy:
    - check: "recognized_revenue_usd = 0 OR order_status = 'delivered'"
      description: "Revenue recognized only for delivered orders"

# SLA
sla:
  availability: "99.5%"
  latency: "Data available by 07:00 UTC daily"

# Versioning policy
versioning:
  breaking_changes_require: "30 days notice + migration guide"
  backwards_compatible_changes: "immediate, no notice required"
  deprecation_policy: "6 months notice before removal"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

-- macros/enforce_contract.sql
{% macro enforce_contract(contract_path) %}
    -- Auto-generate dbt tests from contract YAML
    {% set contract = fromyaml(load_file(contract_path)) %}
    {% for column in contract.schema %}
        {% if column.nullable == false %}
            -- {{ column.name }}: not_null
        {% endif %}
    {% endfor %}
{% endmacro %}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

# models/marts/core/_core_models.yml
models:
  - name: fct_orders
    config:
      contract:
        enforced: true  # dbt will fail if the model's schema doesn't match
    columns:
      - name: order_id
        data_type: varchar
        constraints:
          - type: not_null
          - type: unique
      - name: total_amount_usd
        data_type: numeric
        constraints:
          - type: not_null

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

# scripts/check_contract_breaking_change.py
"""
Run in CI to detect breaking schema changes.
Compares current contract with production schema.
"""
import json
import sys
import subprocess

def get_production_schema(model_name: str) -> dict:
    """Query information_schema for current column types"""
    result = subprocess.run([
        "dbt", "run-operation", "get_columns_in_relation",
        "--args", f'{{"relation_name": "{model_name}"}}'
    ], capture_output=True, text=True)
    return json.loads(result.stdout)

def check_breaking_changes(contract_path: str, model_name: str) -> list[str]:
    """Returns list of breaking changes found"""
    import yaml

    with open(contract_path) as f:
        contract = yaml.safe_load(f)

    prod_schema = get_production_schema(model_name)
    prod_columns = {col["column"]: col["data_type"] for col in prod_schema}

    breaking_changes = []

    for col in contract["schema"]:
        col_name = col["name"]

        # Column removed
        if col_name not in prod_columns:
            breaking_changes.append(f"BREAKING: Column '{col_name}' removed from production")
            continue

        # Type changed incompatibly
        contract_type = col["type"].upper()
        prod_type = prod_columns[col_name].upper()
        if contract_type != prod_type:
            # Some type changes are safe (VARCHAR widening), others aren't
            if not is_compatible_type_change(prod_type, contract_type):
                breaking_changes.append(
                    f"BREAKING: Column '{col_name}' type changed "
                    f"from {prod_type} to {contract_type}"
                )

    return breaking_changes

if __name__ == "__main__":
    changes = check_breaking_changes("contracts/fct_orders.yaml", "fct_orders")
    if changes:
        print("Breaking contract changes detected:")
        for change in changes:
            print(f"  {change}")
        sys.exit(1)
    print("No breaking changes detected.")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

# great_expectations_suite.py
import great_expectations as gx

context = gx.get_context()

# Define expectation suite
suite = context.add_expectation_suite("fct_orders.critical")

# Connect to data source
datasource = context.sources.add_pandas_filesystem(
    name="warehouse",
    base_directory="/data/exports"
)

asset = datasource.add_csv_asset("fct_orders", batching_regex=r"fct_orders_(?P<date>\d{8})\.csv")

batch_request = asset.build_batch_request()
validator = context.get_validator(batch_request=batch_request, expectation_suite=suite)

# Define expectations
validator.expect_column_values_to_not_be_null("order_id")
validator.expect_column_values_to_be_unique("order_id")
validator.expect_column_values_to_not_be_null("total_amount_usd")
validator.expect_column_values_to_be_between(
    column="total_amount_usd",
    min_value=0,
    max_value=50000
)
validator.expect_column_values_to_be_in_set(
    column="order_status",
    value_set=["pending", "processing", "shipped", "delivered", "cancelled", "refunded"]
)
validator.expect_table_row_count_to_be_between(
    min_value=10000,  # at least 10k rows
    max_value=10_000_000  # no more than 10M (sanity check)
)

# Row-level expectations
validator.expect_column_pair_values_a_to_be_greater_than_b(
    column_A="updated_at",
    column_B="created_at"
)

# Volume anomaly detection (z-score based)
validator.expect_column_stdev_to_be_between(
    column="total_amount_usd",
    min_value=10,  # some variation expected
    max_value=500  # not too much variation
)

validator.save_expectation_suite()

# Run checkpoint and get results
checkpoint = context.add_or_update_checkpoint(
    name="daily_orders_check",
    validations=[{"batch_request": batch_request, "expectation_suite_name": "fct_orders.critical"}]
)

result = checkpoint.run()
if not result.success:
    # Send alert
    send_slack_alert(f"Data quality check failed: {result.statistics}")
    raise ValueError("Data quality check failed — blocking pipeline")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

pip install elementary-data

# Add to packages.yml:
# - package: elementary-data/elementary
#   version: "0.13.0"

dbt deps

# Generate elementary models
dbt run --select elementary

# Run monitoring
edr monitor --slack-webhook ${SLACK_WEBHOOK_URL}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

# models/staging/stg_orders.yml — with elementary monitors
models:
  - name: stg_orders
    meta:
      elementary:
        timestamp_column: created_at
    columns:
      - name: total_amount_usd
        meta:
          elementary:
            monitors:
              - type: anomaly_detection
                anomaly_sensitivity: 3.0  # z-score threshold
              - type: schema_changes

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

-- Create a quality metrics table
-- Run this as part of your dbt pipeline

-- models/monitoring/data_quality_metrics.sql
with metrics as (

    select
        'fct_orders'                    as model_name,
        current_timestamp               as checked_at,
        count(*)                        as row_count,
        count(distinct order_id)        as unique_order_ids,
        sum(case when order_id is null then 1 else 0 end) as null_order_ids,
        sum(case when total_amount_usd < 0 then 1 else 0 end) as negative_amounts,
        min(created_at)                 as earliest_record,
        max(created_at)                 as latest_record,
        datediff('hour', max(created_at), current_timestamp) as hours_since_latest

    from {{ ref('fct_orders') }}

)

select
    *,
    case
        when null_order_ids > 0         then 'FAIL: null PKs'
        when unique_order_ids != row_count then 'FAIL: duplicate PKs'
        when negative_amounts > 0       then 'FAIL: negative amounts'
        when hours_since_latest > 24    then 'WARN: stale data'
        else 'PASS'
    end as quality_status

from metrics

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

# models/marts/core/_core_models.yml
models:
  - name: fct_orders
    description: |
      ## Orders Fact Table

      The core orders fact table. One row per order. This is the **source of truth**
      for all order-related metrics across the organization.

      ### Usage Notes
      - Use `recognized_revenue_usd` for financial reporting (accrual basis)
      - Use `total_amount_usd` for customer-facing revenue figures
      - `order_status` follows the state machine documented in [Confluence](https://your-confluence/order-states)

      ### Grain
      One row per `order_id`.

      ### Refresh
      Updated daily at 06:30 UTC from production PostgreSQL via CDC.

      ### Known Limitations
      - Orders from before 2020-01-01 may have incomplete item data
      - Cancelled orders before 2022-03-15 are incorrectly showing `total_amount_usd = 0`
        (tracked in [DATA-1234](https://jira/DATA-1234))
    columns:
      - name: order_id
        description: "Globally unique order identifier. Format: `ord_[alphanumeric]`"
      - name: recognized_revenue_usd
        description: |
          Revenue recognized per US GAAP accrual rules.
          Equal to `total_amount_usd` for delivered orders, 0 for all others.
          **Do not use for bookings or ARR calculations** — use `total_amount_usd` for those.

1
2
3
4
5
6

dbt docs generate
dbt docs serve  # local preview at http://localhost:8080

# Deploy to S3/GCS for team access
dbt docs generate --no-compile
aws s3 sync target/ s3://your-bucket/dbt-docs/

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

# datahub-ingestion.yaml
source:
  type: dbt
  config:
    manifest_path: /opt/dbt/target/manifest.json
    catalog_path: /opt/dbt/target/catalog.json
    sources_path: /opt/dbt/target/sources.json
    target_platform: snowflake
    target_platform_instance: prod
    # Auto-ingest ownership from dbt YAML
    enable_meta_mapping: true
    meta_mapping:
      owner:
        match: ".*"
        operation: "add_owner"
        config:
          owner_type: group

sink:
  type: datahub-rest
  config:
    server: http://datahub-gms:8080

---
# Also ingest from Snowflake directly for lineage
source:
  type: snowflake
  config:
    account_id: your-snowflake-account
    warehouse: COMPUTE_WH
    role: DATAHUB_ROLE
    include_table_lineage: true
    include_column_lineage: true

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

# models/metrics/_metrics.yml
semantic_models:
  - name: orders
    defaults:
      agg_time_dimension: order_date
    model: ref('fct_orders')
    entities:
      - name: order
        type: primary
        expr: order_id
      - name: user
        type: foreign
        expr: user_id
    dimensions:
      - name: order_date
        type: time
        type_params:
          time_granularity: day
        expr: date_trunc('day', created_at)
      - name: order_status
        type: categorical
      - name: user_country
        type: categorical
    measures:
      - name: order_count
        agg: count_distinct
        expr: order_id
      - name: revenue
        agg: sum
        expr: recognized_revenue_usd
      - name: avg_order_value
        agg: average
        expr: total_amount_usd

metrics:
  - name: monthly_revenue
    type: simple
    label: "Monthly Revenue"
    type_params:
      measure: revenue
    filter: |
      {{ Dimension('order__order_status') }} = 'delivered'

  - name: revenue_growth_mom
    type: derived
    label: "Revenue Growth MoM"
    type_params:
      expr: (revenue - lag_revenue) / lag_revenue
      metrics:
        - name: revenue
        - name: revenue
          offset_window: 1 month
          alias: lag_revenue