Unlocking Data Insights: Maestro 1.4.0 Introduces Custom Metrics for Enhanced Pipeline Monitoring

The recent release of Maestro version 1.4.0 brings a significant enhancement to pipeline orchestration with the inclusion of custom metrics tracking. While tracking execution time and success rates was already possible, users now have the capability to answer specific queries such as how many rows were processed or how many API calls were made during pipeline execution. This feature not only adds depth to the data insights but also enhances the operational oversight of data workflows.

For those unfamiliar with Maestro, it’s a pipeline orchestration package that can be utilized in various data processing environments. You can explore more about its capabilities here.

To install Maestro, simply execute the following command in R:

install.packages("maestro")

Using update_metric() to Record Metrics

Metrics can be any quantifiable measurements, such as the number of rows processed, the volume of data read, or the amount of API calls made. To record a metric, you need to use the update_metric() function within your pipeline, specifying the metric name and its respective value. Maestro links these metrics to the currently executing pipeline and run, enabling comprehensive monitoring.

It's designed so that these metrics are recorded separately and do not affect the return value of the pipeline, ensuring a clean separation of concerns. For example, consider a simple Directed Acyclic Graph (DAG) where data is extracted and subsequently loaded, recording metrics along the way:

#' ./pipelines/metric_example.R
#' @maestroFrequency 1 hour
extract_flights <- function() {
# Simulates fetching 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))
}

Accessing Metrics with get_metrics()

Once the scheduled pipeline runs, metrics can be retrieved using the get_metrics() function. This function returns a data frame containing each recorded metric, along with the respective pipeline name and run ID, providing a clear picture of performance across executions.

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 vPnJoC rows_read 5
2 extract_flights vPnJoC bytes_read 1648
3 load_flights ulLl5m rows_written 5

Each execution of a pipeline is associated with a unique run_id, and metrics from different runs remain separate. This format aids in analyzing trends and monitoring the health of data pipelines over time.

Understanding Execution-Based Metrics

It's vital to recognize that metrics are tied to individual executions. Every time a pipeline runs, a new run_id is generated, with metrics recorded distinctly. If the same metric name is used within a single execution, calling update_metric() again will overwrite the previous value (the most recent value takes precedence).

Re-running the schedule will illustrate this point, as it retains both sets of metrics associated with their respective 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 vPnJoC rows_read 5
2 extract_flights vPnJoC bytes_read 1648
3 extract_flights bilN7s rows_read 5
4 extract_flights bilN7s bytes_read 1648
5 load_flights ulLl5m rows_written 5
6 load_flights R9iMP0 rows_written 5

Accumulating Metrics with Incremental Updates

For cases where metrics need to accumulate over the execution of a pipeline, you can set the parameter increment = TRUE. This allows you to append values rather than overwrite them, such as counting API calls made during loop iterations or tracking retries of a task. If the metric is not yet set during the current run, the first invocation establishes its starting point. The feature accommodates negative values for decrement purposes as well.

#' ./pipelines/metric_increment.R
#' @maestroFrequency 1 day
fetch_paginated <- function() {
results <- list()
for (page in 1:3) {
attempts <- 0
repeat {
  attempts <- attempts + 1
  update_metric("attempts", 1, increment = TRUE)
  # Simulates an API call that may fail
  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)
update_metric("rows_read", nrow(df))
df
}

In this case, both api_calls and attempts will accumulate, giving a single value per pipeline execution. For further details, check out the metrics vignette.

Other Recent Updates in Version 1.4.0

If you're catching up on the updates from the 1.3.x releases, a couple of notable changes include:

• The newly introduced @maestroCascadeTags tag, which allows the propagation of selected metadata (like label, flags, and/or loglevel) across all downstream pipelines in a DAG.
• Improvements to get_slot_usage(), which now includes min_datetime and max_datetime parameters for setting a defined time window instead of anchoring to the current time.

For a complete list of changes and enhancements, refer to the release notes.

As you explore these new features, they're designed to make your orchestration experience more insightful and effective.