Skip to content

Instantly share code, notes, and snippets.

@MaxKrog

MaxKrog/elt-full.sql

Created January 31, 2022 12:56
Download ZIP

Revisions

  1. MaxKrog created this gist Jan 31, 2022.
    71 changes: 71 additions & 0 deletions elt-full.sql
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,71 @@

    with source_data as (
    select * from `company-a.source.source_data`
    ),

    reading_fields as (
    select
    JSON_VALUE(_loaded_data, "$.call_time") as call_time,
    JSON_VALUE(_loaded_data, "$.number") as caller_number,
    JSON_VALUE(_loaded_data, "$.type") as call_type,
    JSON_VALUE(_loaded_data, "$.outcome") as call_outcome,
    JSON_VALUE(_loaded_data, "$.seconds") as call_duration,
    _loaded_at,
    _loaded_data

    from
    source_data
    ),

    cleaned_fields as (
    select
    -- upstream call_time is string representing a timestamp in CET
    parse_timestamp("%Y-%m-%dT%H:%M:%S", call_time, 'Europe/Stockholm') as ct,
    caller_number,
    case
    when call_type = "Incoming" then 'incoming'
    when call_type = "Outgoing" then 'outgoing'
    end as call_type,
    case
    when call_outcome = "Answered" then 'answered'
    when call_outcome = "Unanswered" then 'unanswered'
    end as call_outcome,
    cast(call_duration as float64) * 60 as call_duration,
    from
    reading_fields

    ),

    clean_with_surr_key as (
    select

    {{ dbt_utils.surrogate_key(['call_time', 'caller_number', 'caller_number', 'call_type', 'call_outcome', 'call_duration']) }} as call_surr_key,
    row_number() over (
    partition by {{ dbt_utils.surrogate_key(['call_time', 'caller_number', 'caller_number', 'call_type', 'call_outcome', 'call_duration']) }}
    order by
    _loaded_at desc
    ) end as call_surr_key_rn,
    call_time,
    caller_number,
    call_type,
    call_outcome,
    call_duration,
    _loaded_at,
    _loaded_data

    from
    source_data
    ),

    out as (
    select
    *
    except
    (call_surr_key_rn, _loaded_at, _loaded_data)
    from
    clean_with_surr_key
    where
    call_surr_key_rn = 1
    )

    select * from out order by call_time desc