Functions, purrr, and Reproducible Pipelines

A script that runs once on your laptop is not a pipeline. This lesson is the four things that turn it into one: functions that generalise, iteration that returns values, failures that do not stop the batch, and an environment that is the same next year.

Functions that take column names

The obvious version does not work:

suppressPackageStartupMessages(library(tidyverse))

orders <- tibble(
  channel = c("web","app","web","phone","app","web"),
  country = c("GB","US","GB","NL","US","GB"),
  status  = c("completed","completed","returned","completed","pending","completed"),
  amount  = c(25.50, 12.00, 40.00, 8.75, 63.20, 19.99)
)

summarise_by <- function(data, group_col, value_col) {
  data |> summarise(total = sum(value_col), n = n(), .by = group_col)
}

summarise_by(orders, channel, amount)

Error in `summarise()`:
ℹ In argument: `total = sum(value_col)`
Caused by error:
! object 'value_col' not found

dplyr evaluated value_col as a column name literally. {{ }} — “embrace” — says use what the caller passed:

summarise_by <- function(data, group_col, value_col) {
  data |>
    summarise(total = sum({{ value_col }}), n = n(), .by = {{ group_col }})
}

print(summarise_by(orders, channel, amount))
print(summarise_by(orders, country, amount))

# A tibble: 3 × 3
  channel total     n
  <chr>   <dbl> <int>
1 web      85.5     3
2 app      75.2     2
3 phone     8.75    1

# A tibble: 3 × 3
  country total     n
  <chr>   <dbl> <int>
1 GB      85.5      3
2 US      75.2      2
3 NL       8.75     1

Now the function is a dplyr verb. Name the output column after the input with := and englue():

summarise_by <- function(data, group_col, value_col) {
  data |>
    summarise(
      "{{ value_col }}_total" := sum({{ value_col }}),
      n = n(),
      .by = {{ group_col }}
    )
}
print(summarise_by(orders, channel, amount))

# A tibble: 3 × 3
  channel amount_total     n
  <chr>          <dbl> <int>
1 web            85.5      3
2 app            75.2      2
3 phone           8.75     1

For several columns, ... passes them straight through:

summarise_many <- function(data, ..., value_col) {
  data |> summarise(total = sum({{ value_col }}), n = n(), .by = c(...))
}
print(summarise_many(orders, channel, country, value_col = amount))

# A tibble: 4 × 4
  channel country total     n
  <chr>   <chr>   <dbl> <int>
1 web     GB      85.5      3
2 app     US      75.2      2
3 phone   NL       8.75     1
4 app     GB       0        0

Validate inputs

load_orders <- function(path, min_rows = 1L) {
  stopifnot(
    "path must be a single string" = is.character(path) && length(path) == 1,
    "file does not exist" = file.exists(path)
  )

  data <- readr::read_csv(path, show_col_types = FALSE)

  required <- c("order_id", "amount", "status")
  missing <- setdiff(required, names(data))
  if (length(missing)) {
    cli::cli_abort("Missing required column{?s}: {.field {missing}}")
  }
  if (nrow(data) < min_rows) {
    cli::cli_abort("Expected at least {min_rows} row{?s}, got {nrow(data)}")
  }
  data
}

load_orders("missing.csv")

Error in load_orders("missing.csv") : file does not exist

writeLines(c("order_id,total", "1001,25.50"), "wrong_schema.csv")
load_orders("wrong_schema.csv")

Error in `load_orders()`:
! Missing required columns: `amount` and `status`

cli_abort() handles pluralisation (column vs columns) and formatting, and the message names exactly what is wrong. A pipeline function should fail on its own preconditions rather than producing a confusing error three steps later.

purrr instead of loops

files <- c("orders_jan.csv", "orders_feb.csv", "orders_mar.csv")
for (f in files) writeLines(c("order_id,amount", "1001,25.50", "1002,12.00"), f)

# the loop
all_data <- list()
for (f in files) all_data[[f]] <- read_csv(f, show_col_types = FALSE)
combined_loop <- bind_rows(all_data, .id = "source")

# the map
combined <- map(files, \(f) read_csv(f, show_col_types = FALSE)) |>
  set_names(files) |>
  list_rbind(names_to = "source")

print(combined)
print(identical(dim(combined_loop), dim(combined)))

# A tibble: 6 × 3
  source         order_id amount
  <chr>             <dbl>  <dbl>
1 orders_jan.csv     1001   25.5
2 orders_jan.csv     1002   12  
3 orders_feb.csv     1001   25.5
4 orders_feb.csv     1002   12  
5 orders_mar.csv     1001   25.5
6 orders_mar.csv     1002   12  
[1] TRUE

Same result, and the map version is an expression — there is no partially-filled all_data to reason about if it fails halfway.

The typed variants are the real reason to prefer purrr:

print(map_dbl(list(1:5, 6:10, 11:15), sum))
print(map_chr(files, \(f) as.character(file.size(f))))
print(map_lgl(files, file.exists))

map_dbl(list(1:5, "not a number"), sum)

[1]  15  40  65
[1] "34" "34" "34"
[1] TRUE TRUE TRUE

Error in `map_dbl()`:
ℹ In index: 2.
Caused by error in `sum()`:
! invalid 'type' (character) of argument

map_dbl guarantees a double vector, so a wrong type fails at the element that caused it, with the index. A loop appending to a list would have produced a mixed list and failed somewhere else entirely.

Two inputs at once, and iteration with the index:

thresholds <- c(web = 20, app = 30, phone = 10)

print(map2_dbl(names(thresholds), thresholds,
               \(ch, t) sum(orders$amount[orders$channel == ch & orders$amount > t])))

print(imap_chr(thresholds, \(t, ch) str_glue("{ch}: threshold {t}")))

[1] 65.49 63.20  0.00
                    web                     app                   phone 
   "web: threshold 20"     "app: threshold 30"   "phone: threshold 10" 

walk() is map() for side effects — it returns its input invisibly, so it chains:

walk(files, \(f) cat("processing", f, "\n"))

processing orders_jan.csv 
processing orders_feb.csv 
processing orders_mar.csv 

Failures that do not stop the batch

read_maybe <- function(path) {
  if (!file.exists(path)) stop("no such file: ", path)
  read_csv(path, show_col_types = FALSE)
}

paths <- c("orders_jan.csv", "does_not_exist.csv", "orders_mar.csv")
map(paths, read_maybe)

Error in `map()`:
ℹ In index: 2.
Caused by error in `read_maybe()`:
! no such file: does_not_exist.csv

One bad file and two good ones were wasted. safely() captures instead:

safe_read <- safely(read_maybe)
results <- map(paths, safe_read)

ok     <- keep(results, \(r) is.null(r$error)) |> map("result")
failed <- keep(results, \(r) !is.null(r$error))

cat("succeeded:", length(ok), " failed:", length(failed), "\n")
walk2(paths[map_lgl(results, \(r) !is.null(r$error))], map(failed, \(r) conditionMessage(r$error)),
      \(p, m) cat("  FAILED", p, "-", m, "\n"))

combined <- list_rbind(ok)
cat("rows loaded:", nrow(combined), "\n")

succeeded: 2  failed: 1 
  FAILED does_not_exist.csv - no such file: does_not_exist.csv 
rows loaded: 4 

Two files processed, one reported by name. That is what a batch job should do — partial progress plus an explicit list of what needs attention, rather than all-or-nothing.

possibly() is the shorter form when you only need a default:

sizes <- map_dbl(paths, possibly(\(p) file.size(p), otherwise = NA_real_))
print(sizes)

[1] 34 NA 34

And insistently() retries transient failures:

fetch <- insistently(\(url) httr2::request(url) |> httr2::req_perform(),
                     rate = purrr::rate_backoff(pause_base = 1, max_times = 3))

Testing the functions

# tests/testthat/test-summarise.R
library(testthat)

test_that("summarise_by totals by group", {
  d <- tibble(g = c("a","a","b"), v = c(1, 2, 10))
  out <- summarise_by(d, g, v)

  expect_s3_class(out, "tbl_df")
  expect_equal(nrow(out), 2)
  expect_equal(out$v_total[out$g == "a"], 3)
})

test_that("load_orders rejects a missing column", {
  writeLines(c("order_id,total", "1,2"), tmp <- tempfile(fileext = ".csv"))
  expect_error(load_orders(tmp), "Missing required column")
})

test_that("empty input returns an empty result, not an error", {
  d <- tibble(g = character(), v = numeric())
  expect_equal(nrow(summarise_by(d, g, v)), 0)
})

Rscript -e 'testthat::test_dir("tests/testthat")'

✔ | F W  S  OK | Context
✔ |          3 | summarise                                                    

══ Results ═══════════════════════════════════════════════════════════════
[ FAIL 0 | WARN 0 | SKIP 0 | PASS 3 ]

The third test is the one worth copying — empty input is the case that reaches production untested, because nobody thinks to try it.

Assert on the data too, not only the code:

validate_orders <- function(data) {
  problems <- c(
    if (anyDuplicated(data$order_id)) "duplicate order_id",
    if (any(is.na(data$amount))) str_glue("{sum(is.na(data$amount))} null amounts"),
    if (any(data$amount < 0, na.rm = TRUE)) "negative amounts",
    if (!all(data$status %in% c("completed","returned","refunded","pending")))
      str_glue("unexpected status: {paste(setdiff(unique(data$status), c('completed','returned','refunded','pending')), collapse=', ')}")
  )
  if (length(problems)) cli::cli_abort("Validation failed: {problems}")
  invisible(data)
}

bad <- tibble(order_id = c(1L, 1L), amount = c(25.5, -3), status = c("completed", "unknown"))
validate_orders(bad)

Error in `validate_orders()`:
! Validation failed: duplicate order_id, negative amounts, and unexpected status: unknown

All three problems in one message, so one run tells you everything to fix.

Pinning the environment

Rscript -e 'renv::init()'

* Initializing project ...
* Discovering package dependencies ... Done!
* Copying packages into the cache ... [47/47] Done!
The following package(s) will be updated in the lockfile:

# CRAN -----------------------------------------------------------------
- dplyr         [* -> 1.1.4]
- ggplot2       [* -> 3.5.1]
- tidyr         [* -> 1.3.1]
...
* Lockfile written to '~/bookshop/renv.lock'.

Rscript -e 'renv::snapshot()'
Rscript -e 'renv::status()'

* The project is already synchronized with the lockfile.

On another machine, or next year:

Rscript -e 'renv::restore()'

The following package(s) will be installed:
- dplyr   [1.1.4]
- ggplot2 [3.5.1]
...
Do you want to proceed? [Y/n]: Y
* Installing dplyr [1.1.4] ... OK

Commit renv.lock. Without it a dependency update changes results silently — and in analysis code you find out from the numbers rather than from an error, which is the worst way.

Record the environment in the output as well:

cat("R:", R.version.string, "\n")
cat("run at:", format(Sys.time(), tz = "UTC"), "UTC\n")
print(map_chr(c("dplyr","tidyr","ggplot2"), \(p) as.character(packageVersion(p))) |>
      set_names(c("dplyr","tidyr","ggplot2")))

R: R version 4.5.1 (2026-06-14) 
run at: 2026-09-10 08:14:02 UTC
  dplyr   tidyr ggplot2 
"1.1.4" "1.3.1" "3.5.1" 

A pipeline that skips unchanged work

# _targets.R
library(targets)
tar_option_set(packages = c("dplyr", "readr", "ggplot2"))

list(
  tar_target(raw_file, "data/orders.csv", format = "file"),
  tar_target(orders, read_csv(raw_file, show_col_types = FALSE)),
  tar_target(validated, validate_orders(orders)),
  tar_target(daily, summarise_by(validated, ordered_at, amount)),
  tar_target(plot, ggplot(daily, aes(ordered_at, amount_total)) + geom_line()),
  tar_target(report, ggsave("daily.png", plot, width = 8, height = 4), format = "file")
)

Rscript -e 'targets::tar_make()'

▶ dispatched target raw_file
● completed target raw_file [0.002 seconds]
▶ dispatched target orders
● completed target orders [0.184 seconds]
▶ dispatched target validated
● completed target validated [0.011 seconds]
▶ dispatched target daily
● completed target daily [0.042 seconds]
▶ dispatched target plot
● completed target plot [0.221 seconds]
▶ dispatched target report
● completed target report [0.402 seconds]
▶ ended pipeline [1.104 seconds]

Run it again with nothing changed:

✔ skipped target raw_file
✔ skipped target orders
✔ skipped target validated
✔ skipped target daily
✔ skipped target plot
✔ skipped target report
✔ skipped pipeline [0.088 seconds]

Everything skipped. Change one thing and only what depends on it re-runs:

# edit the plot target only
Rscript -e 'targets::tar_make()'

✔ skipped target raw_file
✔ skipped target orders
✔ skipped target validated
✔ skipped target daily
▶ dispatched target plot
● completed target plot [0.216 seconds]
▶ dispatched target report
● completed target report [0.398 seconds]
▶ ended pipeline [0.702 seconds]

targets hashes each target’s inputs and code, so the dependency graph is derived rather than declared — the same idea as ref() in dbt, applied to R objects. On a pipeline with a 40-minute model fit, this is the difference between iterating on a plot in seconds and in an hour.

Rscript -e 'targets::tar_visnetwork()'
Rscript -e 'targets::tar_read(daily)' 

# A tibble: 90 × 3
  ordered_at amount_total     n
  <date>            <dbl> <int>
1 2026-01-01         289.    12
2 2026-01-02         312.    14
# ℹ 88 more rows

tar_read() pulls any intermediate out of the store without re-running anything, which is how you debug a pipeline step without a script full of saveRDS calls.

Running it unattended

#!/usr/bin/env Rscript
# run_pipeline.R
suppressPackageStartupMessages(library(optparse))

opts <- OptionParser() |>
  add_option("--date", type = "character", default = as.character(Sys.Date() - 1),
             help = "logical date to process [default: yesterday]") |>
  add_option("--dry-run", action = "store_true", default = FALSE) |>
  parse_args()

run_date <- as.Date(opts$date)
cli::cli_inform("Processing {run_date}{if (opts$`dry-run`) ' (dry run)' else ''}")

result <- tryCatch({
  data <- load_orders(str_glue("data/orders_{run_date}.csv"))
  validate_orders(data)
  if (!opts$`dry-run`) write_results(data, run_date)
  list(status = "success", rows = nrow(data))
}, error = function(e) {
  cli::cli_alert_danger("Pipeline failed: {conditionMessage(e)}")
  list(status = "failed", error = conditionMessage(e))
})

jsonlite::write_json(c(result, list(run_date = as.character(run_date),
                                    finished_at = format(Sys.time(), tz = "UTC"))),
                     str_glue("logs/run_{run_date}.json"), auto_unbox = TRUE)

if (result$status == "failed") quit(status = 1)

Rscript run_pipeline.R --date 2026-01-04

ℹ Processing 2026-01-04
✔ Wrote 4,812 rows to daily_revenue

Rscript run_pipeline.R --date 2026-01-05; echo "exit: $?"

ℹ Processing 2026-01-05
✖ Pipeline failed: file does not exist
exit: 1

The three things a scheduler needs: a parameter for the window (lesson 4 of the data engineering track — never Sys.Date() inside the logic), a non-zero exit code on failure, and a machine-readable run record. Without the exit code, cron reports success on every run regardless of what happened.

0 5 * * *  cd /opt/bookshop && Rscript run_pipeline.R >> logs/cron.log 2>&1

Practice

Error in `summarise()`:
! object 'value_col' not found

Error in `map_dbl()`:
ℹ In index: 2.
! invalid 'type' (character) of argument

# map()
Error: no such file: does_not_exist.csv

# safely()
succeeded: 2  failed: 1
  FAILED does_not_exist.csv - no such file

✔ skipped target orders
✔ skipped target daily
▶ dispatched target plot
● completed target plot [0.216 seconds]

That closes the R track. The thread through all ten lessons: the data frame is the unit of work, the same dplyr code serves analysis and pipelines, and everything hard — types, timezones, missing values, leakage, package versions — is a decision made explicit rather than left to a default.