fix: otimiza memória em br_ans_beneficiario e remove chamada duplicada em rf_cnpj

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",
     }

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/"

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(

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"}

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",