From 730c2dd344b4010977be244d312fbebf5c98667f Mon Sep 17 00:00:00 2001 From: Winzen Date: Thu, 13 Aug 2026 06:42:01 -0300 Subject: [PATCH 1/2] =?UTF-8?q?fix(br=5Fans=5Fbeneficiario):=20otimiza=20m?= =?UTF-8?q?em=C3=B3ria=20e=20corrige=20upload=20de=20parquet?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A migração pro poll.py que este PR propunha não é mais necessária: o #1783 já corrigiu poll_source_for_update via compare_against, e br_ans_beneficiario já usa esse default corrigido. Mantido só o que continua valendo independente do mecanismo de poll: - RAW_COLLUNS_TYPE: colunas de texto categóricas (baixa/média cardinalidade) trocadas de str para category — reduz bastante o footprint do DataFrame sem mudar o valor persistido no parquet. - MODALIDADE_OPERADORA recasteada para category depois do remove_accents; del df + gc.collect() por estado em parquet_partition — sem isso a memória de cada estado se acumulava até o gc.collect() do loop de fora, já causou OOM num arquivo pequeno logo depois de um grande. - source_format="parquet" nos dois upload_to_gcs: crawler_ans grava .parquet, e sem declarar o formato o dump_header procurava .csv e não achava nada. - job_variables={"memory": "3Gi"}: pico medido em produção após a otimização foi ~1.78Gi; 3Gi dá ~1.7x de margem. - compare_against="coverage" explícito no poll (já era o comportamento via default desde o #1783; deixado explícito por consistência com os outros 26 flows). Também exclui models/world_aiddata_gcdf/code do Pyrefly (mesmo padrão do br_tse_eleicoes/us_harvard_cbdb/us_cfpb_hmda: pacote .py com imports relativos ao cwd, não notebook) — estava quebrando o type check na main, sem relação com esta mudança. --- .../crawler/ans_beneficiario/constants.py | 45 +++++++++++-------- pipelines/crawler/ans_beneficiario/utils.py | 15 ++++++- .../datasets/br_ans_beneficiario/flows.py | 10 +++++ pyproject.toml | 4 ++ 4 files changed, 54 insertions(+), 20 deletions(-) diff --git a/pipelines/crawler/ans_beneficiario/constants.py b/pipelines/crawler/ans_beneficiario/constants.py index 23b253cff2..91e33a47a8 100644 --- a/pipelines/crawler/ans_beneficiario/constants.py +++ b/pipelines/crawler/ans_beneficiario/constants.py @@ -10,27 +10,36 @@ class constants(Enum): Constant values for the br_ans_beneficiario project """ + # Colunas de texto são majoritariamente categóricas de baixa/média + # cardinalidade (UF, sexo, faixa etária, modalidade, município, plano) + # repetidas em milhões de linhas por arquivo. `category` deduplica os + # valores em vez de guardar um `str` do Python por linha — corta bastante + # o footprint do DataFrame em memória, sem mudar o valor persistido no + # parquet (Arrow grava a string real via dicionário; BigQuery/dbt leem + # como STRING normalmente). `#ID_CMPT_MOVEL` fica de fora porque não há + # coluna com esse nome exato no CSV (o dtype nunca casa e o pandas infere + # sozinho) — não mexemos nisso aqui. RAW_COLLUNS_TYPE = { "#ID_CMPT_MOVEL": str, - "CD_OPERADORA": str, - "NM_RAZAO_SOCIAL": str, - "NR_CNPJ": str, - "MODALIDADE_OPERADORA": str, - "SG_UF": str, - "CD_MUNICIPIO": str, - "NM_MUNICIPIO": str, - "TP_SEXO": str, - "DE_FAIXA_ETARIA": str, - "DE_FAIXA_ETARIA_REAJ": str, - "CD_PLANO": str, - "TP_VIGENCIA_PLANO": str, - "DE_CONTRATACAO_PLANO": str, - "DE_SEGMENTACAO_PLANO": str, - "DE_ABRG_GEOGRAFICA_PLANO": str, - "COBERTURA_ASSIST_PLAN": str, - "TIPO_VINCULO": str, + "CD_OPERADORA": "category", + "NM_RAZAO_SOCIAL": "category", + "NR_CNPJ": "category", + "MODALIDADE_OPERADORA": "category", + "SG_UF": "category", + "CD_MUNICIPIO": "category", + "NM_MUNICIPIO": "category", + "TP_SEXO": "category", + "DE_FAIXA_ETARIA": "category", + "DE_FAIXA_ETARIA_REAJ": "category", + "CD_PLANO": "category", + "TP_VIGENCIA_PLANO": "category", + "DE_CONTRATACAO_PLANO": "category", + "DE_SEGMENTACAO_PLANO": "category", + "DE_ABRG_GEOGRAFICA_PLANO": "category", + "COBERTURA_ASSIST_PLAN": "category", + "TIPO_VINCULO": "category", "QT_BENEFICIARIO_ATIVO": int, "QT_BENEFICIARIO_ADERIDO": int, "QT_BENEFICIARIO_CANCELADO": int, - "DT_CARGA": str, + "DT_CARGA": "category", } diff --git a/pipelines/crawler/ans_beneficiario/utils.py b/pipelines/crawler/ans_beneficiario/utils.py index 22206a8dca..971818f9f7 100644 --- a/pipelines/crawler/ans_beneficiario/utils.py +++ b/pipelines/crawler/ans_beneficiario/utils.py @@ -136,8 +136,12 @@ def parquet_partition(path): df["ano"] = time_col.dt.year df["mes"] = time_col.dt.month - df["MODALIDADE_OPERADORA"] = df["MODALIDADE_OPERADORA"].apply( - remove_accents + # volta pra category depois do apply (remove_accents devolve str + # puro) — mantém a coluna leve para o fatiamento em to_partitions. + df["MODALIDADE_OPERADORA"] = ( + df["MODALIDADE_OPERADORA"] + .apply(remove_accents) + .astype("category") ) df = df.rename( columns={ @@ -163,4 +167,11 @@ def parquet_partition(path): log("Partição feita.") + # Sem isso, a memória de cada estado se acumula até o gc.collect() + # do loop de fora em crawler_ans, que só roda depois dos 27 + # arquivos — já causou OOM num arquivo pequeno (AP) logo depois + # de processar um grande (MG). + del df + gc.collect() + return "/tmp/data/br_ans_beneficiario/output/" diff --git a/pipelines/datasets/br_ans_beneficiario/flows.py b/pipelines/datasets/br_ans_beneficiario/flows.py index fb44dbe73f..6ad0383639 100644 --- a/pipelines/datasets/br_ans_beneficiario/flows.py +++ b/pipelines/datasets/br_ans_beneficiario/flows.py @@ -59,6 +59,7 @@ def br_ans_beneficiario__informacao_consolidada( source_max_date=file_last_date, env="prod", date_format="%Y-%m", + compare_against="coverage", ) if not has_new_data: print(f"Não há atualizações para a tabela {table_id}!") @@ -84,12 +85,16 @@ def br_ans_beneficiario__informacao_consolidada( output_filepath = crawler_ans(files=files) + # crawler_ans -> parquet_partition grava .parquet. Sem declarar o + # formato, o dump_header chamado por upload_to_gcs procura .csv (default) + # e não encontra nada. upload_to_gcs( data_path=output_filepath, dataset_id=dataset_id, table_id=table_id, bucket_name="basedosdados-dev", dump_mode="append", + source_format="parquet", ) run_dbt( @@ -109,6 +114,7 @@ def br_ans_beneficiario__informacao_consolidada( table_id=table_id, bucket_name="basedosdados", dump_mode="append", + source_format="parquet", ) run_dbt( @@ -141,3 +147,7 @@ def br_ans_beneficiario__informacao_consolidada( br_ans_beneficiario__informacao_consolidada.deploy_schedules = [ {"cron": "0 21 * * *", "timezone": "America/Sao_Paulo"} ] +# Pico medido em produção após otimizar parquet_partition (category dtype + +# del/gc.collect() por estado): ~1.78Gi. ~1.7x de margem sobre esse valor. +# pyrefly: ignore [missing-attribute] +br_ans_beneficiario__informacao_consolidada.job_variables = {"memory": "3Gi"} diff --git a/pyproject.toml b/pyproject.toml index bb1fb1c3a6..30d6c2a0b7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -225,6 +225,10 @@ project-excludes = [ # duckdb, with same-dir imports), so duckdb/common/clean/download are # unresolvable from the repo root. "models/us_cfpb_hmda/code", + # Same policy: world_aiddata_gcdf ETL is standalone `.py` (architecture/, + # clean.py, gen_dbt.py, upload.py use importlib.util.spec_from_file_location + # with cwd-relative paths), unresolvable from the repo root. + "models/world_aiddata_gcdf/code", "models/br_tse_eleicoes/code/[[]dbt[]]br_tse_eleicoes.ipynb", "models/world_wb_mides/code/licitacao_item.ipynb", "models/world_olympedia_olympics/code/[[]code[]]world_olympedia_olympics.ipynb", From defac52759dd01a72fc05180dae4f4bd72e564ec Mon Sep 17 00:00:00 2001 From: Winzen Date: Thu, 13 Aug 2026 06:56:40 -0300 Subject: [PATCH 2/2] fix(rf_cnpj): remove chamada duplicada de commit_source_update_task MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit O #1783 moveu commit_source_update_task pra logo após o poll, removendo a chamada antiga do fim do flow em todos os arquivos afetados. O #1798, mergeado depois mas cortado de uma base anterior a esse merge, ainda tinha essa chamada antiga e só editou parâmetros dela (date_format). O merge do git combinou as duas sem detectar a duplicação lógica — commit_source_update_task passou a ser chamado duas vezes por run. Sem impacto de dados (é idempotente), só uma escrita redundante. --- pipelines/crawler/rf_cnpj/flows.py | 9 --------- 1 file changed, 9 deletions(-) diff --git a/pipelines/crawler/rf_cnpj/flows.py b/pipelines/crawler/rf_cnpj/flows.py index a8a8957f7e..cd82c5693c 100644 --- a/pipelines/crawler/rf_cnpj/flows.py +++ b/pipelines/crawler/rf_cnpj/flows.py @@ -192,15 +192,6 @@ def _run_rf_cnpj( bq_project="basedosdados", ) - if folder_date is not None: - commit_source_update_task( - dataset_id=dataset_id, - table_id=table_id, - source_max_date=folder_date, - env="prod", - date_format=DateFormat.YEAR_MONTH, - ) - # estabelecimentos: atualiza diretório de empresas if table_id == "estabelecimentos": run_dbt(