Skip to content

DBT: Global Coffee 2020 External

File location: s3://trase-storage/world/global_coffee/global_coffee_2020_external.parquet

DBT model name: global_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_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
country_of_production not_null_global_coffee_2020_external_country_of_production — 2026-10-19 06:23 pass 🔍 run query
country_of_production relationships_global_coffee_2020_external_country_of_production__country_name__ref_postgres_countries_ — 2026-10-19 06:24 pass 🔍 run query
exporter_group trader_group_check_global_coffee_2020_external_exporter_group__year — 2026-10-19 06:25 pass 🔍 run query
mass_tonnes dbt_utils_expression_is_true_global_coffee_2020_external_mass_tonnes___0 — 2026-10-19 06:19 pass 🔍 run query
year accepted_values_global_coffee_2020_external_year__2020 — 2026-10-19 06:12 pass 🔍 run query

Not referenced by any model or exposure.

Models

No called script or script source not found.

"""
Shareable version of the Global Coffee 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("global_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_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("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