From d325296fb852327bdfc51f3b1ca3b653a556f736 Mon Sep 17 00:00:00 2001 From: Davi Cavalcante <93160711+DaviMacielCavalcante@users.noreply.github.com> Date: Thu, 6 Aug 2026 17:28:24 -0300 Subject: [PATCH 1/3] =?UTF-8?q?fix(br=5Fme=5Fcomex=5Fstat):=20migra=20para?= =?UTF-8?q?=20o=20poll=20de=20cobertura=20e=20destrava=20a=20atualiza?= =?UTF-8?q?=C3=A7=C3=A3o=20mensal?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pipelines/datasets/br_me_comex_stat/README.md | 243 ++++++++++++++++++ pipelines/datasets/br_me_comex_stat/flows.py | 45 ++-- 2 files changed, 265 insertions(+), 23 deletions(-) create mode 100644 pipelines/datasets/br_me_comex_stat/README.md diff --git a/pipelines/datasets/br_me_comex_stat/README.md b/pipelines/datasets/br_me_comex_stat/README.md new file mode 100644 index 0000000000..7d93839b60 --- /dev/null +++ b/pipelines/datasets/br_me_comex_stat/README.md @@ -0,0 +1,243 @@ +# Documentação do Conjunto de Dados: Comex Stat (Estatísticas de Comércio Exterior) + +Este documento registra o contexto e as decisões da base do Comex Stat, para quem +for mexer no pipeline depois. + +--- + +## Sobre a fonte + +O Comex Stat é a base de estatísticas de comércio exterior do MDIC, montada a +partir das declarações de importação e exportação processadas pela Receita +Federal. + +- [Página de base de dados bruta (é o que o flow raspa)](https://www.gov.br/mdic/pt-br/assuntos/comercio-exterior/estatisticas/base-de-dados-bruta) +- [Diretório dos arquivos CSV](https://balanca.economia.gov.br/balanca/bd/comexstat-bd) + +Quatro tabelas, dois recortes: + +| Tabela | Arquivo na fonte | Grão | Partições | +|---|---|---|---| +| `ncm_exportacao` | `ncm/EXP_.csv` | NCM × país × UF × via × URF | `ano`, `mes` | +| `ncm_importacao` | `ncm/IMP_.csv` | idem | `ano`, `mes` | +| `municipio_exportacao` | `mun/EXP__MUN.csv` | SH4 × país × município | `ano`, `mes`, `sigla_uf` | +| `municipio_importacao` | `mun/IMP__MUN.csv` | idem | `ano`, `mes`, `sigla_uf` | + +As quatro são `part_bdpro` com `free_lag` de 6 meses (o padrão de `PartBdpro`): +os últimos 6 meses ficam atrás do BD Pro e o resto é livre. A janela rola +sozinha a cada run, via `sync_table_coverage_task`. + +## Particularidades que mandam no desenho do pipeline + +### A fonte publica o ano inteiro, sempre + +Cada arquivo é o ano corrente completo, não o mês novo. Três consequências, todas +boas: + +1. **Não existe backfill a fazer.** Qualquer run refaz janeiro–mês corrente. Se o + flow perdeu um mês, a run seguinte traz o mês perdido junto — não precisa de + loop de meses como o `br_me_caged` nem de listagem de FTP como o CNES. +2. **É idempotente.** `to_partitions` grava `data.csv` com nome fixo por partição + (`pipelines/utils/utils.py`), então re-subir o ano sobrescreve os mesmos + objetos no GCS. `dump_mode="append"` não duplica. +3. **Revisão de mês passado entra de graça.** O MDIC revisa meses anteriores + dentro do mesmo arquivo anual, e como o ano inteiro é reprocessado, a revisão + chega sem tratamento especial. + +Na virada de ano o `parse_last_date` devolve `YYYY-01` e o download passa a +buscar só o arquivo novo — dezembro do ano anterior já está materializado das +runs anteriores. + +### A data da fonte vem de um `

` raspado, não dos arquivos + +`parse_last_date` (`pipelines/crawler/me_comex_stat/tasks.py`) lê o seletor +`#parent-fieldname-text > h2` da página de base bruta: + +```text +'Últimos dados disponíveis: janeiro - julho de 2026' → '2026-07' +``` + +Duas fragilidades para ter em mente: + +- O texto passa por um dicionário `{"julho de 2026": "2026-07", ...}`. Se o MDIC + mudar a redação do título, é `KeyError` — falha alta, não silenciosa, o que é + o comportamento desejado. +- **Esse título fala pelas quatro tabelas ao mesmo tempo.** Se um dia os arquivos + de município passarem a ter defasagem maior que os de NCM, o gate vai liberar + as tabelas de município todo dia sem nada novo para trazer (a run é inofensiva, + mas reprocessa e roda dbt à toa). Em 2026-08-06 conferi mês a mês os quatro + arquivos: todos com `CO_MES` até 7, sem divergência. Se aparecer churn diário + em `municipio_*`, é aqui que se olha. + +### A validação contra os totais oficiais derruba a run + +`validate_table` (`pipelines/crawler/me_comex_stat/utils.py`) baixa o arquivo de +conferência do MDIC (`*_TOTAIS_CONFERENCIA*.csv`) e compara, para o ano de +referência, as somas de `valor_fob_dolar`, `peso_liquido_kg` e +`quantidade_estatistica` mais a contagem de linhas. Qualquer divergência levanta +`ValueError`. + +Isso roda dentro do clean, **antes de qualquer upload**, então uma falha nunca +suja o BigQuery. E como o `Table.Update` só avança no fim do flow, a run seguinte +tenta de novo sozinha. O cenário esperado é o MDIC publicar o arquivo de dados +antes de atualizar o de conferência: a run do dia quebra, a do dia seguinte passa. + +### Uma única fonte para as quatro tabelas + +As quatro tabelas apontam para o mesmo `RawDataSource` no backend +(`347bae65-…`, a página de base bruta). Ou seja, o `RawDataSource.Update` é um +**ponteiro compartilhado**: a primeira das quatro que rodar no dia o avança, e as +outras três encontram ele já em dia. + +Isso não atrapalha o gate, porque o gate compara contra o `Table.Update`, que é +por tabela. Mas explica uma coisa que confunde ao olhar o site: as quatro exibem +a mesma "última atualização na fonte original" mesmo tendo coberturas diferentes. + +--- + +## Pipeline: a armadilha que já congelou este conjunto + +### O gate antigo comparava data de cobertura com relógio + +Até 2026-08, o flow usava `poll_source_for_update_task`, cujo gate é: + +```python +source_max_date > client.get_table_update_latest(...) +``` + +O lado esquerdo é uma **data de cobertura** (`2026-07-01` = "a fonte tem dados até +julho"). O lado direito é um **relógio**: `register_table_materialization_task` +grava ali o `bq.last_modified()`, o horário da última materialização. + +Como o dado do mês M sai no começo de M+1, a data de cobertura fica sempre cerca +de um mês atrás do relógio. O gate só passa quando faz mais de um mês desde a +última materialização — na prática, o conjunto ingeria **mês sim, mês não**. + +Estado em 2026-08-06, que é o retrato exato do problema: + +| Tabela | `Table.Update` (relógio) | `RawSource.Update` (cobertura) | Cobertura publicada | +|---|---|---|---| +| `ncm_exportacao` | 2026-07-03 | 2026-06-01 | …2026-06 | +| `ncm_importacao` | 2026-07-03 | 2026-06-01 | …2026-06 | +| `municipio_exportacao` | 2026-06-26 | 2026-06-01 | …2026-05 | +| `municipio_importacao` | 2026-06-26 | 2026-06-01 | …2026-05 | + +Com julho na fonte, a conta do gate ficava `2026-07-01 > 2026-07-03` para NCM — +falso, travado. As tabelas de município já tinham perdido junho pelo mesmo +motivo (`2026-06-01 > 2026-06-26` é falso), e é por isso que a cobertura delas +parou em maio. + +O sintoma no Prefect é traiçoeiro: a run termina `COMPLETED`, o `Poll` fica em +dia e o log diz apenas `Não há novas atualizações na fonte original`. Uma +pipeline morta é indistinguível de uma saudável pelo estado da run. + +### O modelo novo, adotado aqui em 2026-08 + +O flow passou para o modelo de reconciliação de cobertura +(`pipelines/utils/metadata/poll.py`), o mesmo do CNES e do +`br_me_caged`: + +```python +register_source_coverage_task(...) # grava o Poll e avança RawSource.Update +check_source_is_ahead_of_table_task(...) # gate: RawSource.Update > Table.Update +... +sync_table_coverage_task(...) # cobertura + RAPs + Table.Update como cobertura +``` + +Agora os dois lados da comparação são data de cobertura. O `commit_source_update_task` +sumiu: quem grava o ponteiro da fonte é o `register_source_coverage_task`, no +começo do flow. O commit deferido deixou de ser necessário porque o gate passou a +depender do `Table.Update`, que só avança no `sync_table_coverage_task`, no fim — +uma falha no meio não trava a run seguinte. + +O `register_source_coverage_task` fica **fora** do `if not force_run`: o Poll é +gravado em toda run, inclusive nas forçadas. Só o gate é pulado. + +### Migrar exige semear o `Table.Update` — não é opcional + +Esta é a parte que congelou o CNES de 2026-06-17 a 2026-08-03 e que o diff do +`br_me_caged` não cobre. O `Table.Update` que já está gravado é **relógio**, e o +gate novo o interpreta como cobertura. Sem semear, ele compara +`2026-07-01 > 2026-07-03` e devolve falso para sempre — o modelo novo herda o +congelamento do antigo. + +Nada no código valida isso nem corrige sozinho: o único escritor do valor certo +(`sync_table_coverage_task`) está **depois** do gate que o valor errado bloqueia. + +Duas rotas equivalentes: + +```python +# 1. semear a cobertura já materializada +from pipelines.utils.metadata.client import MetadataClient +import datetime + +client = MetadataClient(env="prod") +seeds = { + "ncm_exportacao": datetime.date(2026, 6, 1), + "ncm_importacao": datetime.date(2026, 6, 1), + "municipio_exportacao": datetime.date(2026, 5, 1), + "municipio_importacao": datetime.date(2026, 5, 1), +} +for table_id, latest in seeds.items(): + client.upsert_table_update("br_me_comex_stat", table_id, latest=latest) +```text + +``` +# 2. disparar cada deployment uma vez com force_run +parameters = {"force_run": true} +``` + +Conferir depois com `client.get_table_update_latest(...)`: tem que voltar data de +cobertura (dia 1º, meia-noite), nunca timestamp com hora. Se voltar hora +diferente de meia-noite, é relógio e o gate está travado. + +**O seed tem que chegar junto com o deploy do código novo.** Enquanto o flow +antigo estiver no ar, todo run bem-sucedido dele chama +`register_table_materialization_task`, que regrava o `Table.Update` como relógio +— desfazendo o seed. Pior: o seed *destrava* o gate antigo também (ele compara +contra o mesmo campo), então semear e não fazer o deploy provoca exatamente o run +que apaga o seed. Se isso acontecer, é só semear de novo com a cobertura +materializada mais recente. + +--- + +## Ambiente e execução + +Os flows escrevem em **prod por padrão** (`materialize_after_dump=True`, +`update_metadata=True`, `target="prod"`). Uma run disparada com `{}` sobe dado em +prod, reescreve cobertura e reaplica as Row Access Policies. Para testar sem +tocar em prod: + +```json +{"materialize_after_dump": false, "update_metadata": false, "force_run": true} +``` + +Crons (America/Sao_Paulo), em `flows.py`: + +| Flow | Cron | +|---|---| +| `ncm_exportacao`, `ncm_importacao` | `0 8,17 * * 1-5` | +| `municipio_importacao` | `0 20 * * 1-5` | +| `municipio_exportacao` | `0 21 * * 1-5` | + +As tabelas de NCM rodam duas vezes por dia. Com o gate novo, a primeira run que +detectar novidade materializa e a segunda encontra o `Table.Update` já em dia e +encerra. + +### Verificação local não alcança este flow + +`load_flows_from_file` não importa `flows.py` no venv 3.10 do repo, porque +`pipelines/utils/metadata/domain.py` usa `StrEnum` (Python 3.11+). Não é um +defeito deste conjunto — vale para todo flow que importa `domain.py`, CNES e +CAGED inclusive. A imagem de execução é `prefecthq/prefect:3-python3.12`, onde +funciona. Para checar a descoberta localmente, importe com um shim: + +```python +import enum +if not hasattr(enum, "StrEnum"): + class StrEnum(str, enum.Enum): pass + enum.StrEnum = StrEnum +``` + +Lembrando que `load_flows_from_file` **engole ImportError** e devolve `{}` — um +import quebrado vira ausência silenciosa de deploy, não erro. diff --git a/pipelines/datasets/br_me_comex_stat/flows.py b/pipelines/datasets/br_me_comex_stat/flows.py index b450405f67..bfb23baaf1 100644 --- a/pipelines/datasets/br_me_comex_stat/flows.py +++ b/pipelines/datasets/br_me_comex_stat/flows.py @@ -18,9 +18,9 @@ YearMonth, ) from pipelines.utils.metadata.tasks import ( - commit_source_update_task, - poll_source_for_update_task, - register_table_materialization_task, + check_source_is_ahead_of_table_task, + register_source_coverage_task, + sync_table_coverage_task, ) from pipelines.utils.tasks import ( rename_flow_run_dataset_table, @@ -49,16 +49,24 @@ def _flow( last_date = parse_last_date(link=comex_constants.DOWNLOAD_LINK.value) - if not force_run: - has_new_data = poll_source_for_update_task( - dataset_id=dataset_id, - table_id=table_id, - source_max_date=last_date, - env="prod", - date_format="%Y-%m", - ) - if not has_new_data: - return + # A fonte é uma só para as quatro tabelas, então este Update é um + # ponteiro compartilhado: quem rodar primeiro no dia o avança. O gate + # abaixo continua por tabela, porque compara contra o Table.Update. + register_source_coverage_task( + dataset_id=dataset_id, + table_id=table_id, + source_max_date=last_date, + env="prod", + date_format="%Y-%m", + ) + + if not force_run and not check_source_is_ahead_of_table_task( + dataset_id=dataset_id, + table_id=table_id, + env="prod", + ): + print(f"Tabela {table_id} já cobre a fonte — encerrando") + return download_br_me_comex_stat( table_name=table_name, @@ -107,7 +115,7 @@ def _flow( ) if update_metadata: - register_table_materialization_task( + sync_table_coverage_task( dataset_id=dataset_id, table_id=table_id, coverage=PartBdpro( @@ -118,15 +126,6 @@ def _flow( bq_project="basedosdados", ) - if last_date is not None: - commit_source_update_task( - dataset_id=dataset_id, - table_id=table_id, - source_max_date=last_date, - env="prod", - date_format="%Y-%m", - ) - _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}] return _flow From 65de653eb30689b529fb9ae129e8b9d5d27d546d Mon Sep 17 00:00:00 2001 From: Davi Cavalcante <93160711+DaviMacielCavalcante@users.noreply.github.com> Date: Fri, 7 Aug 2026 11:32:53 -0300 Subject: [PATCH 2/3] =?UTF-8?q?fix(br=5Fme=5Fcomex=5Fstat):=20remove=20cha?= =?UTF-8?q?mada=20=C3=B3rf=C3=A3=20de=20commit=5Fsource=5Fupdate=5Ftask?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pipelines/datasets/br_me_comex_stat/flows.py | 9 --------- 1 file changed, 9 deletions(-) diff --git a/pipelines/datasets/br_me_comex_stat/flows.py b/pipelines/datasets/br_me_comex_stat/flows.py index b1d8f14192..ce98875292 100644 --- a/pipelines/datasets/br_me_comex_stat/flows.py +++ b/pipelines/datasets/br_me_comex_stat/flows.py @@ -129,15 +129,6 @@ def _flow( bq_project="basedosdados", ) - if last_date is not None: - commit_source_update_task( - dataset_id=dataset_id, - table_id=table_id, - source_max_date=last_date, - env="prod", - date_format="%Y-%m", - ) - # pyrefly: ignore [missing-attribute] _flow.deploy_schedules = [{"cron": cron, "timezone": "America/Sao_Paulo"}] return _flow From 36e69bea14a103c728aecc29f9025f2f7a32bbbb Mon Sep 17 00:00:00 2001 From: Davi Cavalcante <93160711+DaviMacielCavalcante@users.noreply.github.com> Date: Fri, 7 Aug 2026 15:15:52 -0300 Subject: [PATCH 3/3] =?UTF-8?q?docs(br=5Fme=5Fcomex=5Fstat):=20corrige=20p?= =?UTF-8?q?roblemas=20do=20README=20apontadas=20na=20revis=C3=A3o?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pipelines/datasets/br_me_comex_stat/README.md | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/pipelines/datasets/br_me_comex_stat/README.md b/pipelines/datasets/br_me_comex_stat/README.md index 7d93839b60..bff3ef6655 100644 --- a/pipelines/datasets/br_me_comex_stat/README.md +++ b/pipelines/datasets/br_me_comex_stat/README.md @@ -167,7 +167,7 @@ Nada no código valida isso nem corrige sozinho: o único escritor do valor cert Duas rotas equivalentes: ```python -# 1. semear a cobertura já materializada +# semear a cobertura já materializada from pipelines.utils.metadata.client import MetadataClient import datetime @@ -180,11 +180,12 @@ seeds = { } for table_id, latest in seeds.items(): client.upsert_table_update("br_me_comex_stat", table_id, latest=latest) -```text - ``` -# 2. disparar cada deployment uma vez com force_run -parameters = {"force_run": true} + +Ou, em vez disso, disparar cada deployment uma vez com estes parâmetros: + +```json +{"force_run": true} ``` Conferir depois com `client.get_table_update_latest(...)`: tem que voltar data de