Maestro 1.4.0 Enhances Pipeline Orchestration with Custom Metric Recording

The latest release of the R package maestro, version 1.4.0, introduces a significant enhancement for pipeline orchestration: the ability to record custom metrics. This update addresses a critical gap in previous versions, moving beyond the automatic tracking of execution time and success/failure statuses to allow users to quantify specific aspects of their data processing workflows. With this new functionality, users can now gain deeper insights into pipeline performance, such as the number of rows processed, the volume of data handled, or the frequency of external API calls.

maestro is an established package designed for orchestrating complex data pipelines within the R ecosystem. It facilitates the definition, scheduling, and execution of directed acyclic graphs (DAGs) of tasks, ensuring that data processing workflows are managed efficiently and reliably. For those new to maestro, further details can be found on its official documentation site. The package is readily available for installation from the Comprehensive R Archive Network (CRAN) via the command:

install.packages("maestro")

Recording Custom Metrics with update_metric()

The core of this new feature lies in the update_metric() function. This function enables users to record any quantifiable value that can be represented numerically. Examples include the count of rows processed by a specific stage, the total bytes of data read or written, the number of API requests made by a task, or even the number of retries attempted for a particularly challenging operation.

To implement custom metrics, developers simply call update_metric() from within their pipeline code. The function requires two primary arguments: the name of the metric (a character string) and its corresponding numeric value. maestro then automatically associates this metric with the currently executing pipeline and its specific run instance.

A key design principle of update_metric() is that these metrics are recorded as a side effect, separate from the pipeline’s primary return value. This means they do not become part of the data passed downstream in a DAG, preventing unintended data dependencies or pollution of the main data flow. This separation ensures that the core data processing logic remains clean and focused, while performance and operational metrics are captured independently.

Illustrative Example: Tracking Flight Data Processing

To demonstrate the utility of update_metric(), consider a simplified two-step pipeline designed to extract and load flight data. This example showcases how to record metrics related to data volume at different stages.

#' ./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)
  )

  # Record the number of rows read
  update_metric("rows_read", nrow(df))
  # Record the size of the data frame in bytes
  update_metric("bytes_read", as.numeric(object.size(df)))

  df # Return the extracted data


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

  # Simulates writing data to a file
  write.csv(.input, tempfile(fileext = ".csv"))

  # Record the number of rows written
  update_metric("rows_written", nrow(.input))

In this script, the extract_flights function not only retrieves data but also records the number of rows and the memory footprint (in bytes) of the fetched data frame using update_metric(). The load_flights function, which receives the output from extract_flights, then records the number of rows it successfully writes.

Retrieving and Analyzing Metrics with get_metrics()

Once pipelines have been scheduled and executed, the get_metrics() function provides a consolidated view of all recorded metrics. This function returns a tidy data frame, where each row represents a single metric observation. The output includes crucial metadata such as:

• pipe_name: The name of the pipeline from which the metric originated.
• run_id: A unique identifier for the specific execution instance of the pipeline.
• metric: The name of the recorded metric.
• value: The numeric value of the metric.

This structured output makes it straightforward to analyze pipeline performance over time, identify bottlenecks, or monitor resource consumption.

To illustrate, let’s simulate running a schedule and retrieving its metrics:

library(maestro)

# Build the pipeline schedule
schedule <- build_schedule(quiet = TRUE)

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

# Retrieve all recorded metrics
metrics_data <- get_metrics(schedule)
print(metrics_data)

The expected output would resemble the following:

# 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

It is important to note that each pipeline within a DAG receives its own distinct run_id, even if they are part of the same overall pipeline execution. This granular identification is crucial for detailed analysis. Because the output is a standard data frame, users can easily perform further summarization, aggregation, or export the data to databases or files for long-term monitoring and reporting.

Metrics as Per-Execution Snapshots

A fundamental aspect of maestro‘s metric system is that metrics are tied to individual pipeline executions. Each time a pipeline is run, it is assigned a new run_id, and its associated metrics are stored independently of previous runs. This ensures that historical performance data is preserved and accessible.

Within a single execution of a pipeline, if update_metric() is called multiple times with the same metric name, the latest value will overwrite the previous ones. This "last write wins" behavior is suitable for metrics that represent a final state or a single computed value for that execution.

To demonstrate how historical data is retained, let’s run the schedule a second time and then retrieve the metrics again:

# Run the schedule again
status <- run_schedule(
  schedule,
  orch_frequency = "1 hour",
  quiet = TRUE
)

# Retrieve metrics after the second run
metrics_data_after_second_run <- get_metrics(schedule)
print(metrics_data_after_second_run)

The output will now show metrics from both execution runs, distinguished by their respective run_ids:

# 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

This output clearly shows that metrics from the first run (identified by vPnJoC and ulLl5m) are preserved alongside the metrics from the second run (identified by bilN7s and R9iMP0). This historical data is invaluable for tracking trends, detecting regressions, and understanding the long-term behavior of data pipelines.

Accumulating Metrics with increment = TRUE

In scenarios where a metric is not a single static value but rather accumulates over the course of a pipeline’s execution, the increment = TRUE argument to update_metric() becomes essential. This is particularly useful for tasks that involve loops, retries, or iterative processes. For instance, one might want to count the total number of API calls made within a loop, or the cumulative number of attempts to complete a task that is prone to intermittent failures.

When increment = TRUE, the provided value is added to the metric’s existing value for the current run. If the metric has not yet been set for that run, the first call with increment = TRUE will initialize it with the given value. This feature also supports decrementing by providing a negative value. Crucially, even with accumulation, each metric within a single pipeline run ultimately resolves to a single aggregated value.

Example: Paginated Data Fetching with Retries

Consider a pipeline designed to fetch data from an API that returns results in pages and occasionally fails, requiring retries.

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

  results <- list()

  for (page in 1:3) 
    attempts <- 0

    repeat 
      attempts <- attempts + 1
      # Increment the 'attempts' metric for each retry
      update_metric("attempts", 1, increment = TRUE)

      # Simulate an API call that might fail
      ok <- runif(1) > 0.5
      if (ok 

    # Increment the 'api_calls' metric for each successful page fetch
    update_metric("api_calls", 1, increment = TRUE)
    results[[page]] <- data.frame(page = page, value = rnorm(10))
  

  df <- do.call(rbind, results)

  # For a one-off value like total rows, the default overwrite is fine
  update_metric("rows_read", nrow(df))

  df

In this fetch_paginated function, update_metric("attempts", 1, increment = TRUE) is called within the retry loop. This ensures that the attempts metric accurately reflects the total number of times the API was called, including retries, across all pages. Similarly, update_metric("api_calls", 1, increment = TRUE) counts each successful page fetch. The rows_read metric, however, uses the default behavior (overwriting) to record the total number of rows processed after all pages have been successfully fetched.

The result is that api_calls will end up with a value of 3 (one for each page), while attempts will reflect the total number of retry attempts made across all pages. Each metric still provides a single, consolidated value per pipeline run, making it easy to interpret.

For an in-depth exploration of maestro‘s metrics capabilities, users are directed to the dedicated metrics vignette available at https://whipson.github.io/maestro/articles/maestro-9-metrics.html.

Other Recent Enhancements in maestro

Beyond the introduction of custom metrics, the maestro package has seen several other improvements in its recent 1.3.x releases. These updates, while perhaps less prominent than the metrics feature, contribute to the overall robustness and usability of the package. Key among these include:

• Enhanced Error Handling and Reporting: Improvements have been made to how maestro captures and reports errors that occur during pipeline execution. This allows for more precise debugging and faster resolution of issues.
• Configuration Flexibility: New options have been added to provide users with greater control over pipeline scheduling and execution parameters, enabling finer-tuning of workflow behavior.
• Performance Optimizations: Various internal optimizations have been implemented to improve the speed and efficiency of pipeline runs, particularly for large-scale or complex workflows.

A comprehensive list of all changes and updates can be found in the official release notes for maestro at https://whipson.github.io/maestro/news/index.html.

The ongoing development of maestro, highlighted by the significant addition of custom metric recording in version 1.4.0, underscores the package’s commitment to providing powerful and flexible tools for R users managing complex data processing tasks. The ability to quantify pipeline behavior at a granular level empowers users to optimize performance, monitor resource usage, and ensure the reliability of their data workflows. As the field of data science continues to evolve, such robust orchestration tools become increasingly vital.