Skip to content

DBT: Crushing Facilities 2003 2025

File location: s3://trase-storage/brazil/logistics/abiove/out/CRUSHING_FACILITIES_2003_2025.csv

DBT model name: crushing_facilities_2003_2025

Explore on Metabase: Full table; summary statistics

Explore dependencies/lineage: link

Relies on script: trase/data/brazil/logistics/abiove/out/CRUSHING_FACILITIES_2003_2025.py

Description

This dbt model just raises an error; the actual script that produces the data lives at trase/data/brazil/logistics/abiove/out/CRUSHING_FACILITIES_2003_2025.py, an update of crushing_facilities_2003_2024, following the same pattern of combining that year's ABIOVE installed-capacity survey with the JJ Hinrichsen Anuário.

Capacity figures for facilities missing from ABIOVE's official survey are filled in from the JJ Hinrichsen Anuário (see the jj_hinrichsen_anuario doc below for what that source is and how it's processed). # JJ Hinrichsen Anuário (Brazil crushing capacity)

This dataset is the Brazil oilseed-crushing-capacity table extracted from the Anuário JJ (Yearbook JJ), an annual industry yearbook published by J.J. Hinrichsen S.A., a commodity broker ("Corredor - Broker") based in Buenos Aires, Argentina (www.jotajota.com.ar). The yearbook covers the whole "Southern Cone" oilseed/grain complex, including planting and harvest calendars, crushing/milling/ginning capacity, export statistics, prices, and freight terms for Argentina, Brazil, Paraguay, Bolivia, Uruguay and Chile.

Trase only extracts one table from it: "Brazil (I) Installed crushing capacity for oilseeds", which lists individual crushing plants (company, municipality, state, daily capacity in metric tons/day, process, seed type, remarks), grouped by economic group, and reissued roughly twice a year (H1/H2). That single table is then reused across more than one commodity within Trase — see "How we use the dataset in Trase" below — because several of the economic groups it lists (e.g. JBS, Bunge, Cargill, ADM) operate both crushing and meatpacking/beef facilities. We used to also use the dataset for Argentina, Paraguay, and Bolivia, but we no longer update those.

What the dataset contains

Per plant/row in the source PDF table:

  • Company / economic group (e.g. Bunge, Cargill, ADM, LDC, Caramuru, ...)
  • Plant location: municipality and state (UF)
  • Installed daily crushing capacity, in metric tons/day
  • Process type (e.g. solvent extraction vs. full pressing) and seed(s) crushed (soy, sunseed, peanut, cotton, ...)
  • Operating status (active / stopped / under construction) and free-text remarks (e.g. "Refining 850 t", "Biodiesel")

How we use the dataset in Trase

Brazil's official source for crushing-plant capacity is ABIOVE's (Associação Brasileira das Indústrias de Óleos Vegetais) annual survey "Pesquisa de Capacidade Instalada". The JJ Hinrichsen Anuário is used as a supplementary/cross-check source: where ABIOVE does not report a capacity figure for a facility (or the facility is missing from ABIOVE altogether), the equivalent JJ Hinrichsen figure is used instead, tagged with reference = "Anuário JJ Hinrichsen <year>" in the output dataset.

The combined dataset CRUSHING_FACILITIES_2003_<year>.csv is used downstream by the Brazil soy SEI-PCS models (IndustrialCapacity and IndustrialCapacityFacilities in preparation.py) and by brazil_soy_supply_sheds (CrushingDemand in preparation.py). It is also used as a cross-commodity company/plant/CNPJ reference in Trase's Brazil beef work, for the reason given above (shared economic groups).

Location in S3

All original JJ Hinrichsen PDFs live on S3:

s3://trase-storage/brazil/logistics/crushing_facilities/in/hinrichsen/

Here is a summary of the state of that folder as of August 2026:

Years covered Original PDF(s) in that S3 folder What happens to them
2006-2019 Brazil crushing 2007.pdf, 2010.pdf, 2012.pdf, 2017.pdf; BR_crushingFacilitiesCapacity_2018.pdf, 2019.pdf Hand-compiled into two derived CSVs sitting in the same S3 folder: HINRICHSEN_2007_2017.csv (2006/2007, 2010, 2012, 2017) and br_crushingFacilities_2018_2019.csv (2018, 2019). Both are in turn combined into CRUSHING_FACILITIES_2003_2019.csv (in brazil/logistics/abiove/out/), which feeds the older "v2_6_0"/"v2_6_1" vintages of the Brazil soy dataset. We haven't verified beyond that: whether those vintages are still the live/published dataset or a frozen historical snapshot, and how exactly the ~600-line processing script handles these six PDFs, is untracked as of this doc.
2023 JJHinrichsen2023.pdf Hand-extracted into brazil/logistics/jjhinrichsen/jjhinrichsen_2023.csv, which feeds into CRUSHING_FACILITIES_2003_2023.csv
2025, 2026 JJHinrichsen2025.pdf, JJHinrichsen2026.pdf Hand-transcribed into a hardcoded table inside a Python helper script (not an S3 file — see "The script that is used to process/clean the dataset" below), which feeds into CRUSHING_FACILITIES_2003_2025.csv / a future 2026 equivalent

(For external datasets) How to fetch the data from the source

JJ Hinrichsen S.A. is a commodity broker. It used to physically publish the Anuário as a book and SEI had to pay to have a copy sent to them, but JJ Hinrichsen now makes the Anuário available for free on its website. There is no API, so getting a new edition into Trase is a manual process.

Update frequency: annual (new numbered edition each year; the Brazil crushing-capacity table itself is updated roughly twice a year within an edition, H1/H2).

Step 1: Obtain the new edition

Download the current year's Anuário as a PDF from www.jotajota.com.ar — it's free. Mike is Trase's point of contact for this relationship if there's ever any difficulty.

Step 2: Store the PDF in S3

Upload the PDF to s3://trase-storage/brazil/logistics/crushing_facilities/in/hinrichsen/. Add a small sibling .txt file recording where it came from, who obtained it, when, and why. Add a jjhinrichsen_<year>_pdf entry (referencing this jj_hinrichsen_anuario doc) to trase/data_pipeline/models/_sources_trase_storage.yml so the new edition is visible in dbt lineage.

Step 3: Extract the Brazil crushing-capacity table

There is no automated extraction. For the 2025 and 2026 editions, the process was: manually transcribe the rows of the "Brazil (I) Installed crushing capacity for oilseeds" table with LLM assistance (Google Gemini) to structure each row into a JSON-like record, hardcoded as ABIOVE_CAPACITY_MAP in trase/data/brazil/logistics/abiove/out/helpers/abiove_hinrichsen_capacity.py, matching each row by hand to Trase facility IDs and CNPJ numbers.

The script that is used to process/clean the dataset

ABIOVE_CAPACITY_MAP (produced by hand-transcription, see Step 3 above) is imported and combined with that year's official ABIOVE survey by two year-specific build scripts: CRUSHING_FACILITIES_2003_2024.py (reads ABIOVE_CAPACITY_MAP rows for year == 2024, sourced from the 2023 edition — see "How we use the dataset in Trase" for why the years don't line up 1:1) and CRUSHING_FACILITIES_2003_2025.py (year == 2025, sourced from the 2025 edition). Both: filter ABIOVE_CAPACITY_MAP to active facilities for the target year, fill any still-missing capacities with the median of known capacities, validate coverage against that year's ABIOVE active-facility list, and append the result onto the previous year's combined file (crushing_facilities_2003_<year-1>). Both scripts, and their dbt mock models (crushing_facilities_2003_2024/_2025), now declare their real dbt.ref/dbt.source inputs, so the whole chain from crushing_facilities_2003_2022 through _2025 is visible in dbt lineage — as is crushing_facilities_2003_2023, the (different, non-ABIOVE_CAPACITY_MAP) missing link between them, added at the same time.

One data-entry inconsistency worth flagging: ABIOVE_CAPACITY_MAP has a single row referencing "Anuário JJ Hinrichsen 2024", but no 2024 edition PDF has ever been archived.

Updating for a new edition means manually re-transcribing that year's Brazil table from the new PDF into ABIOVE_CAPACITY_MAP, writing a new CRUSHING_FACILITIES_2003_<year>.py build script, and adding its dbt mock model (dbt.ref the previous year, dbt.source the new jjhinrichsen_<year>_pdf and that year's ABIOVE survey) — there is currently no script that parses the PDF directly, and no downstream model automatically picks up the newest year (see crushing_facilities_2003_2024's doc for who currently reads which year).

History

  • Before 2021: JJ Hinrichsen physically published the Anuário as a book. SEI paid to have a copy sent to them each year. Acquired presumably by Javier.
  • October 2021: JJ Hinrichsen confirmed Trase (via Javier) had already purchased Anuário access through the 2021 edition, covering 2018-2021 crushing capacity for all South American countries — but this wasn't known internally at the time, and delivery of the physical books turned out to require an extra fee.
  • August 2026: Mike downloaded the PDF report from JJ Hinrichsen's website and confirmed that JJ Hinrichsen now makes the Anuário available for free there. Harry uploaded this to S3 and wrote this documentation (with review from Mike and Jailson) and wired it into DBT.

Details

Column Type Description
TRASE_ID VARCHAR Trase municipality identifier for the crushing facility, in the format BR-<IBGE municipality code>.
YEAR VARCHAR
COMPANY VARCHAR
MUNICIPALITY VARCHAR
UF VARCHAR
GEOCODE VARCHAR
CAPACITY VARCHAR
CAPACITY_SOURCE VARCHAR
CNPJ VARCHAR
LAT VARCHAR
LONG VARCHAR
RESOLUTION VARCHAR
CAPACITY_ANNUAL 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
TRASE_ID not_null_crushing_facilities_2003_2025_TRASE_ID — 2026-09-07 06:15 pass 🔍 run query
TRASE_ID relationships_crushing_facilities_2003_2025_TRASE_ID__trase_id__source_trase_storage_raw_municipality_ — 2026-09-07 06:16 pass 🔍 run query

Not referenced by any model or exposure.

Models

Sources

from trase.tools.aws.metadata import write_csv_for_upload
from trase.tools.aws.aws_helpers_cached import get_pandas_df
from helpers.abiove_hinrichsen_capacity import *

import pandas as pd
import numpy as np

"""
    Configuration
"""
YEAR = 2025

PATH_PREVIOUS_FACILITIES = (
    "brazil/logistics/abiove/out/CRUSHING_FACILITIES_2003_2024.csv"
)

PATH_ABIOVE = "brazil/logistics/abiove/ori/Pesquisa-de-Capacidade-Instalada_2025.xlsx"

PATH_MUNICIPALITY = "brazil/dictionaries/municipality.csv"

PATH_OUTPUT = "brazil/logistics/abiove/out/CRUSHING_FACILITIES_2003_2025.csv"

COLS = [
    "TRASE_ID",
    "YEAR",
    "COMPANY",
    "MUNICIPALITY",
    "UF",
    "GEOCODE",
    "CAPACITY",
    "CAPACITY_SOURCE",
    "CNPJ",
    "LAT",
    "LONG",
    "RESOLUTION",
    "CAPACITY_ANNUAL",
]

"""
    Helper functions
"""


def clean_processing(df: pd.DataFrame, year: int):

    for col in df.select_dtypes(include=["object", "string"]).columns:
        df[col] = df[col].str.strip().str.upper()

    df["cnpj"] = df["cnpj"].str.replace(r"\D", "", regex=True).str.zfill(14)

    df["capacity_tn_day"] = (
        df["capacity_tn_day"]
        .astype(str)
        .str.replace(".", "", regex=False)
        .str.replace("O", "0", regex=False)
        .replace("", np.nan)
        .astype(float)
    )

    # Keep only currently active crushing facilities
    df = df.query(f'status == "ATIVA" and year == {year}')

    return df


def enrich_data(df: pd.DataFrame, year: int = 2025):

    # Rename columns to match the standard output schema
    map_cols = {"capacity_tn_day": "capacity", "reference": "capacity_source"}
    df = df.rename(columns=map_cols)

    # Add metadata and derived fields required for the output dataset
    df["year"] = year
    df["geocode"] = df["trase_id"].str.split("-").str[1]
    df["resolution"] = "DISTRICT"
    df["lat"] = np.nan
    df["long"] = np.nan
    df["capacity_annual"] = df["capacity"].mul(
        288
    )  # Assumes approximately 288 operating days per year

    # Standardize column names
    df.columns = [col.upper() for col in df.columns]

    # Return dataset using the predefined output column order
    return df[COLS]


"""
    Pipeline
"""


if __name__ == "__main__":

    # Load source datasets
    # ====================================================================================
    df_previous_years = get_pandas_df(
        key=PATH_PREVIOUS_FACILITIES,
        bucket="trase-storage",
    )

    df_abiove = get_pandas_df(
        key=PATH_ABIOVE,
        bucket="trase-storage",
        xlsx=True,
        sheet_name="3.Unidades de Processamento",
        usecols="B:E,F:G,I",
        header=7,
        skipfooter=7,
    )

    df_municipality = get_pandas_df(key=PATH_MUNICIPALITY, sep=",")

    # Load manually matched ABIOVE and JJ report capacity values.
    # The mapping was created by reconciling facility names and capacities
    # between ABIOVE records and the JJ report PDF.
    # The JSON structure was generated with assistance from Google Gemini.
    df_capacity_abiove_jj = pd.DataFrame(ABIOVE_CAPACITY_MAP).drop_duplicates()

    # Data processing
    # ====================================================================================

    # Clean and standardize capacity reference data
    df_capacity_abiove_jj = clean_processing(df_capacity_abiove_jj, YEAR)

    # Fill missing capacities using estimated values and update the source label
    df_capacity_abiove_jj["reference"] = np.where(
        df_capacity_abiove_jj["capacity_tn_day"].isna(),
        "Estimated",
        df_capacity_abiove_jj["reference"],
    )

    # Replace missing capacity values with the 25th percentile of known capacities
    df_capacity_abiove_jj["capacity_tn_day"] = df_capacity_abiove_jj[
        "capacity_tn_day"
    ].fillna(df_capacity_abiove_jj["capacity_tn_day"].median())

    # Add derived fields and transform data into the output schema
    df_capacity_abiove_jj = enrich_data(df_capacity_abiove_jj)

    # Prepare ABIOVE reference records for validation
    abiove_active_facilities = df_abiove[
        (df_abiove[YEAR] == "Ativa") & (df_abiove["Soja"] == "X")
    ]

    for col in abiove_active_facilities.select_dtypes(
        include=["object", "string"]
    ).columns:
        abiove_active_facilities[col] = (
            abiove_active_facilities[col].str.strip().str.upper()
        )

    # Validation checks
    # ====================================================================================

    # Ensure every active soybean processing company listed by ABIOVE
    # is present in the generated output dataset
    assert (
        abiove_active_facilities["Empresas"]
        .isin(df_capacity_abiove_jj["COMPANY"])
        .all()
    )

    # Ensure every company-municipality combination from ABIOVE
    # is represented in the generated output dataset
    check = abiove_active_facilities.merge(
        df_capacity_abiove_jj[["COMPANY", "MUNICIPALITY", "UF"]],
        left_on=["Empresas", "Município", "UF"],
        right_on=["COMPANY", "MUNICIPALITY", "UF"],
        how="left",
        indicator=True,
    )

    missing = check[check["_merge"] == "left_only"]

    assert missing.empty, (
        "Companies/municipalities in Abiove reference file missing from output:\n"
        f"{missing[['Empresas', 'Município', 'UF']]}"
    )

    # Ensure all trase_ids are in oficial list of municipality ids
    mask_municipalities = df_capacity_abiove_jj["TRASE_ID"].isin(
        df_municipality["trase_id"]
    )

    missing_municipalities = df_capacity_abiove_jj.loc[
        ~mask_municipalities,
        ["TRASE_ID", "MUNICIPALITY", "UF", "YEAR"],
    ].drop_duplicates()

import tempfile
from pathlib import Path

if not missing_municipalities.empty:
    output_path = Path(tempfile.gettempdir()) / "missing_municipalities.csv"
    missing_municipalities.to_csv(output_path, index=False)
    print(f"Missing municipalities written to: {output_path}")

    assert missing_municipalities.empty, (
        "Municipality trase_ids in Abiove output missing from official municipality list:\n"
        f"{missing_municipalities}"
    )

    # Combine historical records with the newly processed 2025 facilities
    # ====================================================================================

    df_crushing_plants = pd.concat(
        [df_previous_years[COLS], df_capacity_abiove_jj[COLS]]
    )

    df_crushing_plants["CNPJ"] = (
        df_crushing_plants["CNPJ"]
        .fillna("00000000000000")
        .astype(int)
        .astype("string")
        .str.replace(r"\D", "", regex=True)
        .str.zfill(14)
    )

    print(" ---------------------------- ")
    print("Process successfully completed!")
    print(" ---------------------------- ")

    # Export final dataset
    # ====================================================================================

    write_csv_for_upload(df=df_crushing_plants, key=PATH_OUTPUT, bucket="trase-storage")

import pandas as pd


def model(dbt, cursor):
    dbt.ref("crushing_facilities_2003_2024")  # previous year's combined file
    dbt.source(
        "trase-storage-raw", "pesquisa-de-capacidade-instalada_2025"
    )  # ABIOVE survey
    dbt.source(
        "trase-storage-raw", "jjhinrichsen_2025_pdf"
    )  # ABIOVE_CAPACITY_MAP source for year 2025 rows
    dbt.source(
        "trase-storage-raw", "municipality"
    )  # validates all TRASE_IDs are known municipalities

    raise NotImplementedError()
    return pd.DataFrame({"hello": ["world"]})