The hardware and bandwidth for this mirror is donated by METANET, the Webhosting and Full Service-Cloud Provider.
If you wish to report a bug, or if you are interested in having us mirror your free-software or open-source project, please feel free to contact us at mirror[@]metanet.ch.

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 LjN5Zs rows_read        5
2 extract_flights LjN5Zs bytes_read    1648
3 load_flights    FihN58 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.

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 LjN5Zs rows_read        5
2 extract_flights LjN5Zs bytes_read    1648
3 extract_flights G6pYC8 rows_read        5
4 extract_flights G6pYC8 bytes_read    1648
5 load_flights    FihN58 rows_written     5
6 load_flights    mUpGEN 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.

These binaries (installable software) and packages are in development.
They may not be fully stable and should be used with caution. We make no claims about them.