Skip to content

DBT: Diet Trase Coffee 2020 External

File location: s3://trase-storage/diet-trase/diet_trase_coffee_2020_external.parquet

DBT model name: diet_trase_coffee_2020_external

Explore on Metabase: Full table; summary statistics

Explore dependencies/lineage: link

Description

Shareable version of the results with a selected set of fields and only including exports.

Details

Column Type Description
year BIGINT
country_of_production VARCHAR
country_of_production_iso2 VARCHAR
port_of_export_name VARCHAR
hs6 VARCHAR
exporter_name VARCHAR
exporter_node_id BIGINT
exporter_group VARCHAR
highest_known_exporter_owner_group VARCHAR
importer_name VARCHAR
importer_group VARCHAR
highest_known_importer_owner_group VARCHAR
country_of_first_import VARCHAR
country_of_first_import_iso2 VARCHAR
country_of_first_import_economic_bloc VARCHAR
production_region_id VARCHAR
production_region VARCHAR
mass_tonnes DOUBLE
mass_tonnes_raw_equivalent DOUBLE
fob DOUBLE

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_trader_groups_diet_trase_coffee_2020_external_exporter_group__2020 — 2026-07-07 15:09 pass 🔍 run query
country_of_production not_null_diet_trase_coffee_2020_external_country_of_production — 2026-07-07 15:12 pass 🔍 run query
country_of_production relationships_diet_trase_coffee_2020_external_country_of_production__country_name__ref_postgres_countries_ — 2026-07-07 15:13 pass 🔍 run query
mass_tonnes dbt_utils_expression_is_true_diet_trase_coffee_2020_external_mass_tonnes___0 — 2026-07-07 15:10 pass 🔍 run query
year accepted_values_diet_trase_coffee_2020_external_year__2020 — 2026-07-07 15:07 pass 🔍 run query

Not referenced by any model or exposure.

Models

No called script or script source not found.

"""
Shareable version of the Diet Trase results with a selected set of fields and only including exports.
"""

import polars as pl


def model(dbt, cursor):
    dbt.config(
        materialized="external",
    )

    lf = dbt.ref("diet_trase_coffee_2020").pl(lazy=True)

    lf = lf.filter(~pl.col("is_domestic"))

    columns = [
        "year",
        "country_of_production",
        "country_of_production_iso2",
        "port_of_export_name",
        "hs6",
        "exporter_name",
        "exporter_node_id",
        "exporter_group",
        "highest_known_exporter_owner_group",
        "importer_name",
        "importer_group",
        "highest_known_importer_owner_group",
        "country_of_first_import",
        "country_of_first_import_iso2",
        "country_of_first_import_economic_bloc",
        "production_region_id",
        "production_region",
        "mass_tonnes",
        "mass_tonnes_raw_equivalent",
        "fob",
    ]

    lf = lf.select(columns)

    # Recode low-volume rows to UNKNOWN trader identities before aggregation.
    low_volume = pl.col("mass_tonnes_raw_equivalent") < 0.18
    lf = lf.with_columns(
        [
            pl.when(low_volume)
            .then(pl.lit("UNKNOWN"))
            .otherwise(pl.col("exporter_name"))
            .alias("exporter_name"),
            pl.when(low_volume)
            .then(pl.lit(15221616))
            .otherwise(pl.col("exporter_node_id"))
            .alias("exporter_node_id"),
            pl.when(low_volume)
            .then(pl.lit("UNKNOWN"))
            .otherwise(pl.col("exporter_group"))
            .alias("exporter_group"),
            pl.when(low_volume)
            .then(pl.lit("UNKNOWN"))
            .otherwise(pl.col("highest_known_exporter_owner_group"))
            .alias("highest_known_exporter_owner_group"),
            pl.when(low_volume)
            .then(pl.lit("UNKNOWN"))
            .otherwise(pl.col("importer_name"))
            .alias("importer_name"),
            pl.when(low_volume)
            .then(pl.lit("UNKNOWN"))
            .otherwise(pl.col("importer_group"))
            .alias("importer_group"),
            pl.when(low_volume)
            .then(pl.lit("UNKNOWN"))
            .otherwise(pl.col("highest_known_importer_owner_group"))
            .alias("highest_known_importer_owner_group"),
            pl.when(low_volume)
            .then(pl.lit("XX"))
            .otherwise(pl.col("production_region_id"))
            .alias("production_region_id"),
            pl.when(low_volume)
            .then(pl.lit("UNKNOWN"))
            .otherwise(pl.col("production_region"))
            .alias("production_region"),
        ]
    )

    # aggregate by mass_tonnes and fob
    group_cols = [
        c
        for c in columns
        if c not in ("mass_tonnes", "mass_tonnes_raw_equivalent", "fob")
    ]
    lf = (
        lf.group_by(group_cols)
        .agg(
            [
                pl.sum("mass_tonnes").alias("mass_tonnes"),
                pl.sum("mass_tonnes_raw_equivalent").alias(
                    "mass_tonnes_raw_equivalent"
                ),
                pl.sum("fob").alias("fob"),
            ]
        )
        .with_columns(
            [
                pl.col("mass_tonnes").round(4),
                pl.col("fob").round(2),
            ]
        )
    )

    return lf