Pipeline Completo com DuckDB

Este guia demonstra o pipeline completo do datacaged: desde a verificação do repositório HuggingFace até consultas analíticas no DuckDB com dplyr e SQL.

1. Verificar disponibilidade antes de baixar

Antes de iniciar qualquer download, verifique se o repositório HuggingFace está acessível e quais competências estão disponíveis.

library(datacaged)

# Verifica se o repositório HuggingFace está online e mede a latência
caged_status()
#> OK Repositório HuggingFace online
#> ℹ Latência: 312 ms
#> ℹ Dataset: https://huggingface.co/datasets/alexsandroprado/caged
# Lista os 12 months mais recentes disponíveis no Novo CAGED
caged_hf_files()

# Lista todas as competências do CAGED antigo
caged_hf_files(type = "antigo", n = Inf)

# Lista competências do CAGED Ajustes
caged_hf_files(type = "ajustes", n = 24)

# Capturar o resultado para uso programático
disponivel <- caged_hf_files(n = 1, verbose = FALSE)
cat("Competência mais recente:", disponivel$competencia, "\n")

2. Download dos microdados

O caged_download() gerencia cache local automaticamente: arquivos já baixados não são re-baixados. O retorno é um manifest — data.frame com o status de cada arquivo.

# Baixa Novo CAGED de 2023 inteiro (MOV + FOR + EXC por competência)
manifest <- caged_download(
  years    = 2023,
  months   = seq_len(12L),
  destdir = "~/dados/caged_cache"   # omitir para usar cache padrão
)

# Inspecionar o manifest
dplyr::count(manifest, type, status)
#> # A tibble: 4 × 3
#>   type  status       n
#>   <chr> <chr>    <int>
#> 1 EXC   baixado     12
#> 2 FOR   baixado     12
#> 3 MOV   baixado     12
#> 4 MOV   cache        0

# Arquivos disponíveis para processar
manifest |>
  dplyr::filter(status %in% c("baixado", "cache")) |>
  dplyr::select(type, competencia, nome_arquivo, arquivo)
# Baixa CAGED antigo — arquivo nacional único por competência
manifest_ant <- caged_download(years = 2018:2019, months = seq_len(12L))

# Baixar também os ajustes do mesmo período
caged_adjustments_load(
  years    = 2018:2019,
  months   = seq_len(12L),
  db_path = "caged_historico.duckdb"
)

3. Parse dos arquivos

O caged_parse() extrai o .7z, lê o .txt e devolve um tibble normalizado.

# Parsear um único arquivo
df_jan <- caged_parse("~/dados/caged_cache/NOVO_CAGED/2023/202301/CAGEDMOV202301.7z")

# Inspecionar
dplyr::glimpse(df_jan)
#> Rows: 3,665,155
#> Columns: 29
#> $ competenciamov      <dbl> 202301, 202301, ...
#> $ uf                  <dbl> 11, 11, ...
#> $ municipio           <dbl> 1100015, ...
#> $ saldomovimentacao   <dbl> 1, -1, ...
#> $ salario             <dbl> 1412.00, 2800.50, ...

# Parsear vários arquivos do mesmo type (devem ser todos MOV, ou todos FOR, etc.)
arquivos_mov <- list.files(
  "~/dados/caged_cache/NOVO_CAGED/2023",
  pattern    = "CAGEDMOV",
  recursive  = TRUE,
  full.names = TRUE
)
df_mov_2023 <- caged_parse_batch(arquivos_mov)

4. Gravar no DuckDB

# Gravar o tibble parseado no banco
caged_to_duckdb(df_mov_2023, db_path = "caged.duckdb")

# Re-runs são seguros: competências já no banco são puladas
caged_to_duckdb(df_mov_2023, db_path = "caged.duckdb")
#> ℹ 12 competências já no banco — pulando.

# Para regravar (ex: após corrigir dados):
caged_to_duckdb(df_mov_2023, db_path = "caged.duckdb",
                overwrite_competencies = TRUE)

# Gravar numa table específica sem autodetectar
caged_to_duckdb(df_jan, db_path = "caged.duckdb", table = "caged_mov")

5. Pipeline completo em um único comando

Para o caso mais comum — baixar tudo e gravar no banco — use caged_load():

library(datacaged)

# ── Novo CAGED 2022-2023 ──────────────────────────────────────────────────────
caged_load(
  years    = 2022:2023,
  months   = seq_len(12L),
  db_path = "caged.duckdb"
)

# ── CAGED antigo 2015-2019 ────────────────────────────────────────────────────
caged_load(
  years    = 2015:2019,
  months   = seq_len(12L),
  db_path = "caged.duckdb"
)

# ── CAGED Ajustes 2015-2019 ───────────────────────────────────────────────────
caged_adjustments_load(
  years    = 2015:2019,
  months   = seq_len(12L),
  db_path = "caged.duckdb"
)

# ── Verificar o banco resultante ──────────────────────────────────────────────
caged_info("caged.duckdb")
#> ── caged.duckdb ──────────────────────────────────────────────────────────
#> Tamanho do arquivo: 4.2 GB
#> ── Tabelas ───────────────────────────────────────────────────────────────
#> * "caged_mov"     Registros: 43,981,860   Competências: 202201 – 202312
#> * "caged_for"     Registros:  1,093,176   Competências: 202201 – 202312
#> * "caged_exc"     Registros:     94,800   Competências: 202201 – 202312
#> * "caged_antigo"  Registros: 28,450,000   Competências: 201501 – 201912
#> * "caged_ajustes" Registros:    342,000   Competências: 201501 – 201912

6. Consultas com dplyr

library(dplyr)

con <- caged_connect("caged.duckdb")
# ── Tabelas disponíveis ───────────────────────────────────────────────────────
DBI::dbListTables(con)

# ── Referência lazy (não carrega na memória) ──────────────────────────────────
mov <- tbl(con, "caged_mov")

# ── Saldo mensal ──────────────────────────────────────────────────────────────
mov |>
  group_by(competenciamov) |>
  summarise(
    admissoes    = sum(saldomovimentacao == 1,  na.rm = TRUE),
    desligamentos = sum(saldomovimentacao == -1, na.rm = TRUE),
    saldo        = sum(saldomovimentacao,        na.rm = TRUE)
  ) |>
  arrange(competenciamov) |>
  collect()

# ── Saldo por UF e setor ──────────────────────────────────────────────────────
mov |>
  filter(competenciamov >= 202301) |>
  group_by(uf, secao) |>
  summarise(saldo = sum(saldomovimentacao, na.rm = TRUE)) |>
  arrange(desc(saldo)) |>
  collect()

# ── Salário médio por escolaridade e sexo ─────────────────────────────────────
mov |>
  filter(!is.na(salario), salario > 0) |>
  group_by(graudeinstrucao, sexo) |>
  summarise(
    salario_medio  = mean(salario, na.rm = TRUE),
    n              = n()
  ) |>
  collect()

# ── Top 10 municípios por admissões ──────────────────────────────────────────
mov |>
  filter(saldomovimentacao == 1) |>
  group_by(municipio) |>
  summarise(admissoes = n()) |>
  slice_max(admissoes, n = 10) |>
  collect()

7. Consultas com SQL direto

# Saldo líquido por seção da CNAE e mês
DBI::dbGetQuery(con, "
  SELECT
    competenciamov                    AS competencia,
    secao,
    SUM(CASE WHEN saldomovimentacao =  1 THEN 1 ELSE 0 END) AS admissoes,
    SUM(CASE WHEN saldomovimentacao = -1 THEN 1 ELSE 0 END) AS desligamentos,
    SUM(saldomovimentacao)                                   AS saldo,
    ROUND(AVG(salario), 2)                                  AS salario_medio
  FROM caged_mov
  WHERE competenciamov >= 202301
  GROUP BY competenciamov, secao
  ORDER BY competenciamov, saldo DESC
")

# Distribuição etária por faixa e type de movimentação
DBI::dbGetQuery(con, "
  SELECT
    CASE
      WHEN idade < 25              THEN 'Até 24 anos'
      WHEN idade BETWEEN 25 AND 34 THEN '25-34 anos'
      WHEN idade BETWEEN 35 AND 44 THEN '35-44 anos'
      WHEN idade BETWEEN 45 AND 54 THEN '45-54 anos'
      ELSE '55 anos ou mais'
    END AS faixa_etaria,
    SUM(CASE WHEN saldomovimentacao =  1 THEN 1 ELSE 0 END) AS admissoes,
    SUM(CASE WHEN saldomovimentacao = -1 THEN 1 ELSE 0 END) AS desligamentos
  FROM caged_mov
  GROUP BY faixa_etaria
  ORDER BY faixa_etaria
")

8. Série histórica: unindo Novo CAGED e CAGED antigo

Para construir séries longas que cruzem o ponto de ruptura de 2020, normalize as colunas antes de empilhar.

# Novo CAGED — colunas a normalizar
novo <- tbl(con, "caged_mov") |>
  select(
    competencia    = competenciamov,
    uf,
    saldo          = saldomovimentacao,
    salario,
    sexo,
    idade,
    escolaridade   = graudeinstrucao
  ) |>
  mutate(serie = "novo")

# CAGED antigo — já tem coluna competencia e saldomovimentacao
antigo <- tbl(con, "caged_antigo") |>
  select(
    competencia,
    uf,
    saldo          = saldomovimentacao,
    salario,
    sexo,
    idade,
    escolaridade
  ) |>
  mutate(serie = "antigo")

# Empilha e agrega
serie_hist <- bind_rows(
  novo   |> group_by(competencia, serie) |> summarise(saldo = sum(saldo, na.rm = TRUE)),
  antigo |> group_by(competencia, serie) |> summarise(saldo = sum(saldo, na.rm = TRUE))
) |>
  collect() |>
  arrange(competencia)

9. Exportar resultados

library(dplyr)

# Exportar para CSV
resultado <- tbl(con, "caged_mov") |>
  group_by(competenciamov, uf) |>
  summarise(saldo = sum(saldomovimentacao, na.rm = TRUE)) |>
  collect()

readr::write_csv(resultado, "saldo_uf_mensal.csv")

# Exportar todas as tabelas do banco para Parquet (nativo DuckDB, muito mais rápido)
caged_to_parquet("caged.duckdb", output_dir = "~/exports")

# Exportar só caged_mov particionado por UF
caged_to_parquet(
  "caged.duckdb",
  output_dir   = "~/exports",
  tables       = "caged_mov",
  partition_by = "uf"
)

# Fechar conexão
DBI::dbDisconnect(con, shutdown = TRUE)

10. Atualização incremental

Use caged_update() para manter o banco atualizado sem re-baixar tudo. A função consulta o banco para descobrir a competência máxima e baixa apenas o que é mais recente no HuggingFace.

# Atualiza o Novo CAGED com os meses ainda não presentes no banco
caged_update(db_path = "caged.duckdb")

# Atualizar todas as séries
caged_update(
  db_path = "caged.duckdb",
  series  = c("novo", "antigo", "ajustes")
)

11. Boas práticas