Lab 3: Medallion dbt Modeling — DHIS2 Case

version: 2
 
sources:
  - name: dhis2_raw
    schema: raw
    tables:
      # aggregate
      - name: dataelement
      - name: dataset
      - name: indicator
      - name: period
      - name: periodtype
      - name: organisationunit
      - name: _orgunitstructure
      - name: categorycombo
      - name: categoryoptioncombo
      - name: categoryoption
      - name: datavalue
      # tracker
      - name: trackedentityinstance
      - name: trackedentityattribute
      - name: trackedentityattributevalue
      - name: program
      - name: programstage
      - name: programinstance
      - name: programstageinstance
      - name: trackedentitydatavalue
      - name: relationship
      - name: relationshiptype

-- models/bronze/aggregate/bronze_datavalue.sql
select
    dataelementid,
    periodid,
    sourceid            as organisationunitid,
    categoryoptioncomboid,
    attributeoptioncomboid,
    cast(value as numeric)   as value,
    storedby,
    lastupdated,
    created
from {{ source('dhis2_raw', 'datavalue') }}

-- models/bronze/aggregate/bronze_dataelement.sql
select
    dataelementid,
    uid,
    name,
    shortname,
    domaintype,     -- AGGREGATE or TRACKER
    valuetype,
    lastupdated
from {{ source('dhis2_raw', 'dataelement') }}

-- models/bronze/aggregate/bronze_organisationunit.sql
select
    organisationunitid,
    uid,
    name,
    parentid,
    path,
    hierarchylevel
from {{ source('dhis2_raw', 'organisationunit') }}

-- models/bronze/tracker/bronze_trackedentityinstance.sql
select
    trackedentityinstanceid,
    uid,
    organisationunitid,
    trackedentitytypeid,
    created,
    lastupdated,
    deleted
from {{ source('dhis2_raw', 'trackedentityinstance') }}

-- models/bronze/tracker/bronze_programinstance.sql
select
    programinstanceid,
    uid,
    trackedentityinstanceid,
    programid,
    organisationunitid,
    enrollmentdate,
    incidentdate,
    status,          -- ACTIVE, COMPLETED, CANCELLED
    lastupdated
from {{ source('dhis2_raw', 'programinstance') }}

-- models/bronze/tracker/bronze_programstageinstance.sql
select
    programstageinstanceid,
    uid,
    programinstanceid,
    programstageid,
    organisationunitid,
    executiondate,
    duedate,
    status,          -- COMPLETED, SCHEDULE, SKIPPED, OVERDUE
    lastupdated
from {{ source('dhis2_raw', 'programstageinstance') }}

-- models/bronze/tracker/bronze_trackedentitydatavalue.sql
select
    programstageinstanceid,
    dataelementid,
    value,
    storedby,
    lastupdated
from {{ source('dhis2_raw', 'trackedentitydatavalue') }}

-- models/silver/aggregate/silver_org_units.sql
select
    o.organisationunitid,
    o.uid                as org_unit_uid,
    o.name               as org_unit_name,
    s.namelevel1         as national_name,
    s.namelevel2         as province_name,
    s.namelevel3         as district_name,
    s.namelevel4         as facility_name,
    o.hierarchylevel
from {{ ref('bronze_organisationunit') }} o
left join {{ source('dhis2_raw', '_orgunitstructure') }} s
    on o.organisationunitid = s.organisationunitid

-- models/silver/aggregate/silver_aggregate_datavalues.sql
select
    dv.dataelementid,
    de.name              as data_element_name,
    dv.organisationunitid,
    dv.periodid,
    p.startdate          as period_start_date,
    p.enddate            as period_end_date,
    dv.categoryoptioncomboid,
    dv.value,
    dv.lastupdated
from {{ ref('bronze_datavalue') }} dv
left join {{ ref('bronze_dataelement') }} de
    on dv.dataelementid = de.dataelementid
left join {{ ref('bronze_period') }} p
    on dv.periodid = p.periodid
where dv.value is not null

-- models/silver/tracker/silver_tracked_entities.sql
select
    tei.trackedentityinstanceid,
    tei.uid              as tracked_entity_uid,
    tei.organisationunitid,
    ou.org_unit_name,
    ou.district_name,
    tei.created          as registration_date,
    tei.deleted
from {{ ref('bronze_trackedentityinstance') }} tei
left join {{ ref('silver_org_units') }} ou
    on tei.organisationunitid = ou.organisationunitid
where tei.deleted = false

-- models/silver/tracker/silver_enrollments.sql
select
    pi.programinstanceid,
    pi.uid                as enrollment_uid,
    pi.trackedentityinstanceid,
    pi.programid,
    prog.name             as program_name,
    pi.organisationunitid,
    pi.enrollmentdate,
    pi.incidentdate,
    pi.status             as enrollment_status
from {{ ref('bronze_programinstance') }} pi
left join {{ ref('bronze_program') }} prog
    on pi.programid = prog.programid

-- models/silver/tracker/silver_events.sql
select
    psi.programstageinstanceid,
    psi.uid                as event_uid,
    psi.programinstanceid,
    psi.programstageid,
    stg.name               as program_stage_name,
    psi.organisationunitid,
    psi.executiondate,
    psi.duedate,
    psi.status             as event_status
from {{ ref('bronze_programstageinstance') }} psi
left join {{ ref('bronze_programstage') }} stg
    on psi.programstageid = stg.programstageid

-- models/silver/tracker/silver_event_data_values.sql
select
    tdv.programstageinstanceid,
    tdv.dataelementid,
    de.name              as data_element_name,
    tdv.value,
    tdv.lastupdated
from {{ ref('bronze_trackedentitydatavalue') }} tdv
left join {{ ref('bronze_dataelement') }} de
    on tdv.dataelementid = de.dataelementid

-- models/gold/aggregate/dim_org_unit.sql
select distinct
    organisationunitid   as org_unit_key,
    org_unit_uid,
    org_unit_name,
    national_name,
    province_name,
    district_name,
    facility_name,
    hierarchylevel
from {{ ref('silver_org_units') }}

-- models/gold/aggregate/dim_period.sql
select distinct
    periodid              as period_key,
    period_start_date,
    period_end_date,
    extract(year from period_start_date)    as year,
    extract(quarter from period_start_date) as quarter,
    extract(month from period_start_date)   as month
from {{ ref('silver_aggregate_datavalues') }}

-- models/gold/aggregate/fact_aggregate_data_value.sql
select
    dataelementid            as data_element_key,
    organisationunitid       as org_unit_key,
    periodid                 as period_key,
    categoryoptioncomboid    as category_option_combo_key,
    value,
    lastupdated
from {{ ref('silver_aggregate_datavalues') }}

-- models/gold/tracker/dim_tracked_entity.sql
select distinct
    trackedentityinstanceid  as tracked_entity_key,
    tracked_entity_uid,
    org_unit_name,
    district_name,
    registration_date
from {{ ref('silver_tracked_entities') }}

-- models/gold/tracker/dim_program.sql
select distinct
    programid   as program_key,
    program_name
from {{ ref('silver_enrollments') }}

-- models/gold/tracker/fact_enrollment.sql
select
    programinstanceid      as enrollment_key,
    trackedentityinstanceid as tracked_entity_key,
    programid              as program_key,
    organisationunitid     as org_unit_key,
    enrollmentdate,
    incidentdate,
    enrollment_status
from {{ ref('silver_enrollments') }}

-- models/gold/tracker/fact_event.sql
select
    programstageinstanceid  as event_key,
    programinstanceid       as enrollment_key,
    programstageid          as program_stage_key,
    organisationunitid      as org_unit_key,
    executiondate,
    duedate,
    event_status
from {{ ref('silver_events') }}

-- models/gold/tracker/fact_event_data_value.sql
select
    programstageinstanceid  as event_key,
    dataelementid           as data_element_key,
    data_element_name,
    value
from {{ ref('silver_event_data_values') }}

# dbt_project.yml
models:
  my_dhis2_project:
    bronze:
      +materialized: view
    silver:
      +materialized: view
    gold:
      +materialized: table
      aggregate:
        fact_aggregate_data_value:
          +materialized: incremental
          +unique_key: [data_element_key, org_unit_key, period_key, category_option_combo_key]
      tracker:
        fact_event_data_value:
          +materialized: incremental
          +unique_key: [event_key, data_element_key]

# models/gold/aggregate/schema.yml
version: 2
 
models:
  - name: dim_org_unit
    columns:
      - name: org_unit_key
        tests: [unique, not_null]
 
  - name: fact_aggregate_data_value
    columns:
      - name: org_unit_key
        tests:
          - relationships:
              to: ref('dim_org_unit')
              field: org_unit_key

# run everything, in dependency order
dbt run
 
# run just the aggregate branch, bronze through gold
dbt run --select path:models/bronze/aggregate+ path:models/silver/aggregate+ path:models/gold/aggregate+
 
# run just the tracker branch
dbt run --select path:models/bronze/tracker+ path:models/silver/tracker+ path:models/gold/tracker+
 
# run one layer at a time (good for a live walkthrough)
dbt run --select bronze.*
dbt run --select silver.*
dbt run --select gold.*
 
# test everything
dbt test