Full Pipeline with DuckDB

This guide demonstrates the full datacaged pipeline: from checking the HuggingFace repository to analytical queries in DuckDB with dplyr and SQL.

1. Check availability before downloading

Antes de iniciar qualquer download, check that the HuggingFace repository is accessible and which competencies are available.

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
# List the 12 most recent competencies available in Novo CAGED
caged_hf_files()

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

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

# Capture result for programmatic use
disponivel <- caged_hf_files(n = 1, verbose = FALSE)
cat("Competência mais recente:", disponivel$competencia, "\n")

2. Download microdata

The caged_download() function manages local cache automatically: already-downloaded files are not re-downloaded. It returns a manifest — a data.frame with the status of each file.

# 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
)

# Inspect 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

# Files available for processing
manifest |>
  dplyr::filter(status %in% c("baixado", "cache")) |>
  dplyr::select(type, competencia, nome_arquivo, arquivo)
# Baixa Legacy CAGED — arquivo nacional único por competência
manifest_ant <- caged_download(years = 2018:2019, months = seq_len(12L))

# Download adjustments for the same period too
caged_adjustments_load(
  years    = 2018:2019,
  months   = seq_len(12L),
  db_path = "caged_historico.duckdb"
)

3. Parse files

caged_parse() extracts the .7z, reads the .txt and returns a normalised tibble.

# Parse a single file
df_jan <- caged_parse("~/dados/caged_cache/NOVO_CAGED/2023/202301/CAGEDMOV202301.7z")

# Inspect
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, ...

# Parse multiple files of the same 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. Write to DuckDB

# Write the parsed tibble to the database
caged_to_duckdb(df_mov_2023, db_path = "caged.duckdb")

# Re-runs are safe: competencies already in the database are skipped
caged_to_duckdb(df_mov_2023, db_path = "caged.duckdb")
#> ℹ 12 competências já no banco — pulando.

# To overwrite (e.g. after correcting data):
caged_to_duckdb(df_mov_2023, db_path = "caged.duckdb",
                overwrite_competencies = TRUE)

# Write to a specific table without auto-detection
caged_to_duckdb(df_jan, db_path = "caged.duckdb", table = "caged_mov")

5. Full pipeline in a single command

For the most common case — download everything and write to the database — use caged_load():

library(datacaged)

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

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

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

# ── Check the resulting database ─────────────────────────────────────────────
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. Queries with dplyr

library(dplyr)

con <- caged_connect("caged.duckdb")
# ── Available tables ─────────────────────────────────────────────────────────
DBI::dbListTables(con)

# ── Lazy reference (does not load into memory) ───────────────────────────────
mov <- tbl(con, "caged_mov")

# ── Monthly balance ───────────────────────────────────────────────────────────
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()

# ── Balance by state and sector ──────────────────────────────────────────────
mov |>
  filter(competenciamov >= 202301) |>
  group_by(uf, secao) |>
  summarise(saldo = sum(saldomovimentacao, na.rm = TRUE)) |>
  arrange(desc(saldo)) |>
  collect()

# ── Average wage by education level and gender ───────────────────────────────
mov |>
  filter(!is.na(salario), salario > 0) |>
  group_by(graudeinstrucao, sexo) |>
  summarise(
    salario_medio  = mean(salario, na.rm = TRUE),
    n              = n()
  ) |>
  collect()

# ── Top 10 municipalities by hires ───────────────────────────────────────────
mov |>
  filter(saldomovimentacao == 1) |>
  group_by(municipio) |>
  summarise(admissoes = n()) |>
  slice_max(admissoes, n = 10) |>
  collect()

7. Direct SQL queries

# Net balance by CNAE section and month
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
")

# Age distribution by range and movement type
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. Historical series: joining Novo CAGED and legacy CAGED

To build long series that cross the 2020 break point, normalise columns before stacking.

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

# Legacy CAGED — 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. Export results

library(dplyr)

# Export to 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")

# Export all database tables to Parquet (native DuckDB, much faster)
caged_to_parquet("caged.duckdb", output_dir = "~/exports")

# Export only caged_mov partitioned by state
caged_to_parquet(
  "caged.duckdb",
  output_dir   = "~/exports",
  tables       = "caged_mov",
  partition_by = "uf"
)

# Close connection
DBI::dbDisconnect(con, shutdown = TRUE)

10. Incremental update

Use caged_update() to keep the database up to date without re-downloading everything. The function queries the database to find the maximum competency and downloads only what is more recent on HuggingFace.

# Update Novo CAGED with months not yet in the database
caged_update(db_path = "caged.duckdb")

# Update all series
caged_update(
  db_path = "caged.duckdb",
  series  = c("novo", "antigo", "ajustes")
)

11. Best practices