Custom Pipeline Metrics

Pipeline metrics are numeric values that track the performance of executed pipelines. Any value that can be represented as a single number can be a metric, but typical metrics are things like the number of rows of a data.frame processed, data volume in bytes, number of API calls, or number of attempts made on a task. Metrics are recorded without being part of the pipeline’s return value, so they never get passed downstream in a DAG.

Other common metrics like execution time and success/fail are already tracked by the orchestrator and are available via get_status() and get_schedule(), so there’s no need to record those yourself.

Recording Metrics

Custom metrics are recorded using the update_metric() function. It is called from within a pipeline function and takes the name of the metric and a numeric value (a single, non-missing number). Maestro automatically attributes the metric to the pipeline (and run) that is currently executing.

#' ./pipelines/metric_example.R
#' @maestroFrequency 1 hour
extract_flights <- function() {

  # Pretends to fetch data from an API
  df <- data.frame(
    flight_id = 1:5,
    airline_code = c("UA", "AC", "AC", "AA", "DE"),
    departure_time = as.POSIXct("2026-10-14 12:00:00") + c(100, 450, 750, 1450, 1750)
  )

  update_metric("rows_read", nrow(df))
  update_metric("bytes_read", as.numeric(object.size(df)))

  df
}

#' @maestroInputs extract_flights
load_flights <- function(.input) {

  write.csv(.input, tempfile(fileext = ".csv"))

  update_metric("rows_written", nrow(.input))
}

After running the schedule, get_metrics() returns a long data.frame (tibble) of every metric recorded: the pipeline it came from (pipe_name), the run it was recorded in (run_id), the metric name, and its numeric value.

library(maestro)

schedule <- build_schedule(quiet = TRUE)

status <- run_schedule(
  schedule,
  orch_frequency = "1 hour",
  quiet = TRUE
)

get_metrics(schedule)

# A tibble: 3 × 4
  pipe_name       run_id metric       value
  <chr>           <chr>  <chr>        <dbl>
1 extract_flights dMaH8w rows_read        5
2 extract_flights dMaH8w bytes_read    1648
3 load_flights    uuP1Tj rows_written     5

Note that each pipeline gets its own run_id, even within a DAG. Since the result is a data.frame, it’s straightforward to summarise further or write out to a database or file.

Note

Metrics are only recorded when a pipeline is executed by run_schedule(). Calling update_metric() while developing or testing a pipeline function interactively is harmless: it emits an informational message and does nothing.

Metrics Are Per Execution

Metrics belong to a single execution of a pipeline. Each time a pipeline runs it gets a new run_id, and its metrics are stored separately from those of earlier runs. Within one execution, calling update_metric() again with the same name overwrites the earlier value (the last write wins).

Running the same schedule a second time shows this. The metrics from both runs are retained and distinguished by run_id.

status <- run_schedule(
  schedule,
  orch_frequency = "1 hour",
  quiet = TRUE
)

get_metrics(schedule)

# A tibble: 6 × 4
  pipe_name       run_id metric       value
  <chr>           <chr>  <chr>        <dbl>
1 extract_flights dMaH8w rows_read        5
2 extract_flights dMaH8w bytes_read    1648
3 extract_flights yTKKHP rows_read        5
4 extract_flights yTKKHP bytes_read    1648
5 load_flights    uuP1Tj rows_written     5
6 load_flights    fjPLUt rows_written     5

Accumulating with increment = TRUE

Sometimes a metric is built up over the course of a pipeline rather than computed once, such as counting API calls inside a loop or the number of attempts on a task. Setting increment = TRUE adds value to the metric’s current value instead of overwriting it. If the metric hasn’t been set yet in the current run, the first call simply sets it. A negative value decrements.

In this example, a pipeline pages through an API and retries a flaky request. Both the number of API calls and the number of attempts accumulate.

#' ./pipelines/metric_increment.R
#' @maestroFrequency 1 day
fetch_paginated <- function() {

  pages <- 1:3
  results <- list()

  for (page in pages) {
    attempts <- 0
    repeat {
      attempts <- attempts + 1
      update_metric("attempts", 1, increment = TRUE)

      # Pretends to call an API that sometimes fails
      ok <- runif(1) > 0.5
      if (ok || attempts >= 3) break
    }

    update_metric("api_calls", 1, increment = TRUE)
    results[[page]] <- data.frame(page = page, value = rnorm(10))
  }

  df <- do.call(rbind, results)

  # Overwriting (default) is fine for a one-off value
  update_metric("rows_read", nrow(df))

  df
}

set.seed(2718)

schedule <- build_schedule(quiet = TRUE)

status <- run_schedule(
  schedule,
  orch_frequency = "1 day",
  quiet = TRUE
)

get_metrics(schedule)

# A tibble: 3 × 4
  pipe_name       run_id metric    value
  <chr>           <chr>  <chr>     <dbl>
1 fetch_paginated SYyFTx attempts      4
2 fetch_paginated SYyFTx api_calls     3
3 fetch_paginated SYyFTx rows_read    30

Each metric has a single value per pipeline run: api_calls ends at 3 (one per page) while attempts is at least 3, depending on how many retries were needed.