Skip to content

DBT: Gold Colombia Cd Coffee 2020

File location: s3://trase-storage/colombia/trade/cd/export/coffee/2020/gold/gold_colombia_cd_coffee_2020.parquet

DBT model name: gold_colombia_cd_coffee_2020

Explore on Metabase: Full table; summary statistics

Explore dependencies/lineage: link

Description

Based on existing cleaned data from colombia/trade/cd/export/2020/CD_COLOMBIA_2020.csv.

Details

Column Type Description
year INTEGER
exporter_label VARCHAR
exporter_node_id BIGINT
exporter_group_name VARCHAR
country_of_destination VARCHAR
mass_tonnes DOUBLE
fob DOUBLE
importer_label VARCHAR
port_of_export_label VARCHAR
hs6 VARCHAR

Review full report including sample errors records if they exist (link)

Test column Test name Failing rows Last test run Last status Query in Metabase
Whole table test check_bol_against_comtrade_gold_colombia_cd_coffee_2020_COL__0901_2101__mass_tonnes__2020 — 2026-07-07 15:08 pass 🔍 run query
Whole table test check_trader_groups_gold_colombia_cd_coffee_2020_exporter_group_name__2020 — 2026-07-07 15:09 pass 🔍 run query
country_of_destination relationships_gold_colombia_cd_coffee_2020_country_of_destination__country_name__ref_postgres_countries_ — 2026-07-07 15:14 pass 🔍 run query

Models

Models

Sources

No called script or script source not found.

-- gold_colombia_cd_coffee_2020
{{ 
    config(materialized='external') 
}}

{% set replace_eu_countries_with_bloc = False %}

-- Using some variables to make it easier to reuse parts of the code (which should then move to a macro)
{% set trader_prefix = "CO-TRADER-" %}
{% set year = 2020 %}


WITH exports AS (
    SELECT
        *,
        UPPER(TRIM(EXPORTER, ', ')) AS exporter_label,
        -- Format the trader trase_id where it exists
        CASE
            WHEN EXPORTER_TAX_ID = 'UNKNOWN' THEN NULL
            ELSE CONCAT('{{ trader_prefix }}', SPLIT_PART(EXPORTER_TAX_ID, '-', 1))
        END AS processed_tax_id
    FROM {{ source('trase-storage-raw', 'cd_colombia_2020') }}
),
-- Expand synonyms for each country in the Trase database
expanded_countries AS (
    SELECT
        country_name,
        country_trase_id,
        UNNEST(synonyms) AS synonym
    FROM {{ ref('postgres_countries') }}
),
-- Join expanded countries and adjust some country names not in the official list
with_official_country_names AS (
    SELECT
        ex.*,
        ec.country_name AS destination_country_name,
        ec.country_trase_id AS destination_country_trase_id
    FROM
        exports ex
    LEFT JOIN expanded_countries ec
        ON ex.COUNTRY_OF_DESTINATION = ec.synonym
),
-- Include the node id and the group name of the traders that are matched by trase_id
with_group_name_by_trase_id AS (
    SELECT 
        wocn.*,
        wocn.processed_tax_id AS trase_id_directly_matched,
        traders.trader_node_id,
        json_extract_string(array_extract(traders.groups, 1), '$.group') AS group_name,
    FROM 
        with_official_country_names AS wocn
    LEFT JOIN 
        {{ ref('postgres_traders') }} AS traders
        ON wocn.processed_tax_id = traders.trase_id
),
-- Take the exporter names still not matched by trase_id
-- we will try to match them by label
exporters_unmatched_by_trase_id AS (
    SELECT
        DISTINCT exporter_label
    FROM with_group_name_by_trase_id
    WHERE group_name IS NULL
),

--------------- THIS SECTION COULD BE REUSABLE --------------
-- Not sure what the best option could be however. Maybe a Python UDF.. 
-- Other option considered was through a macro, but it can't access a previous CTE table. 
-- An option would be to split the first part in an ephemeral model (which doesn't persist 
-- in a table), ref the model within the macro, and then use a materialized model in the end. 
-- But it might be a bit convoluted.

-- Take the info from the traders that have a matching label
filtered_traders_by_label AS (
    SELECT 
        name, 
        trader_node_id,
        trase_id,
        labels,
        UNNEST(groups) AS company_group
    FROM
        {{ ref('postgres_traders') }}
    WHERE EXISTS (
        SELECT 1 FROM exporters_unmatched_by_trase_id
        WHERE list_contains(labels, exporters_unmatched_by_trase_id.exporter_label)
    )
),
-- Flatten the group info from the matching traders, and rank them by relevance
with_traders_group_info AS (
    SELECT 
        name, 
        trader_node_id,
        trase_id,
        json_extract_string(company_group, '$.group')::VARCHAR AS group_name,
        json_extract_string(company_group, '$.time_start')::TIMESTAMP AS time_start,
        json_extract_string(company_group, '$.time_end')::TIMESTAMP AS time_end,
        CASE
            WHEN trase_id LIKE ('{{trader_prefix}}%')
                AND time_start IS NULL AND time_end IS NULL THEN 1
            WHEN trase_id LIKE ('{{trader_prefix}}%') 
                AND YEAR(time_start) <= {{year}} AND YEAR(time_end) >= {{year}} THEN 2
            WHEN trase_id NOT LIKE ('{{trader_prefix}}%')
                AND time_start IS NULL AND time_end IS NULL THEN 3
            WHEN trase_id NOT LIKE ('{{trader_prefix}}%') 
                AND YEAR(time_start) <= {{year}} AND YEAR(time_end) >= {{year}} THEN 4
            WHEN trase_id IS NULL THEN 5
            ELSE NULL
        END AS ranking,
        labels
    FROM filtered_traders_by_label
),
-- Join the export data with the trader info, taking only the top matching ranking when there are multiple matches
-- DISTINCT ON selects the first record per name
filtered_ranked_traders AS (
    SELECT DISTINCT ON (name)
        name, 
        trader_node_id AS trader_node_id_from_label_match,
        trase_id AS trase_id_from_label_match, 
        labels,
        group_name AS group_name_from_label_match,
        time_start,
        time_end,
        ranking
    FROM with_traders_group_info
    ORDER BY name, ranking ASC, trase_id DESC
),
------ END OF POTENTIALY REUSABLE SECTION ------


with_db_trader_info AS (
    SELECT
        with_tr_id.*,
        frt.*
    FROM with_group_name_by_trase_id AS with_tr_id
    LEFT JOIN filtered_ranked_traders AS frt
        ON list_contains(frt.labels, with_tr_id.exporter_label)
),
final_fields AS (
    SELECT
        YEAR::INT AS year,
        exporter_label,
        COALESCE(trader_node_id, trader_node_id_from_label_match) AS exporter_node_id,
        COALESCE(group_name, group_name_from_label_match) AS exporter_group_name,
        destination_country_name AS country_of_destination,
        VOLUME_RAW/1000 AS mass_tonnes,
        FOB AS fob,
        IMPORTER AS importer_label,
        PORT_OF_EXPORT AS port_of_export_label,
        CASE
            WHEN HS6 LIKE '210120' THEN '210111'
            ELSE HS6
        END AS hs6
    FROM with_db_trader_info
)
-- If true (default), replace EU countries with the economic bloc name
-- using 'as_bool' to render the value as a boolean
{% if replace_eu_countries_with_bloc | as_bool %}
,
countries_with_economic_blocs AS (
    SELECT
        country_name,
        UNNEST(economic_bloc) AS economic_bloc
    FROM {{ ref('postgres_countries') }}
),
eu_countries AS (
    SELECT DISTINCT
        country_name AS eu_country,
        'EUROPEAN UNION' AS european_union_country
    FROM countries_with_economic_blocs
    WHERE
        json_extract_string(economic_bloc, '$.economic_bloc') = 'EUROPEAN UNION'
    AND
        country_name != 'UNITED KINGDOM'
),
final_with_eu AS (
    SELECT
        ff.year,
        ff.exporter_label,
        ff.exporter_node_id,
        ff.exporter_group_name,
        COALESCE(eu.european_union_country, ff.country_of_destination) AS country_of_destination,
        ff.mass_tonnes,
        ff.fob,
        ff.importer_label,
        ff.port_of_export_label,
        ff.hs6
    FROM final_fields ff
    LEFT JOIN eu_countries eu
        ON ff.country_of_destination = eu.eu_country
)

SELECT * FROM final_with_eu

{% else %}
-- If not, keep the normal country names

SELECT * FROM final_fields

{% endif %}