maestro 1.4.0 adaugă valori personalizate pentru conducte

Înregistrarea valorilor cu update_metric()

O metrică este orice valoare care poate fi reprezentată ca un singur număr: rânduri procesate, octeți citiți, apeluri API efectuate, încercări de realizare a unei sarcini. Pentru a înregistra unul, sunați update_metric() din interiorul unei conducte cu numele valorii și valoarea acesteia. maestro atribuie automat metrica conductei (și rulării) care se execută în prezent.

Un aspect important al designului este că valorile sunt înregistrate pe lateral. Ele nu fac parte din valoarea de returnare a unei conducte, așa că nu sunt niciodată transmise în aval într-un DAG.

Iată un DAG în doi pași care extrage câteva date de zbor și le încarcă, înregistrând câteva valori pe parcurs:

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

Recuperarea valorilor cu get_metrics()

După rularea programului, get_metrics() returnează un cadru de date lung al fiecărei valori înregistrate. Fiecare rând indică conducta din care provine (pipe_name), rularea în care a fost înregistrată (run_id), cel metric numele și numărul său 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
                       
1 extract_flights vPnJoC rows_read        5
2 extract_flights vPnJoC bytes_read    1648
3 load_flights    ulLl5m rows_written     5

Observați că fiecare conductă își are propria sa run_idchiar și în cadrul unui DAG. Deoarece rezultatul este un data.frame simplu, este ușor să rezumați mai departe sau să scrieți într-o bază de date sau într-un fișier, ceea ce îl face potrivit pentru monitorizarea stării conductei în timp.

Alte câteva lucruri de știut:

• Valorile sunt înregistrate numai atunci când o conductă este executată de run_schedule(). Apel update_metric() în timp ce dezvoltarea sau testarea unei funcții pipeline în mod interactiv este inofensivă: emite un mesaj informațional și nu face nimic.
• Timpul de execuție și succesul/eșecul sunt deja disponibile prin get_status() şi get_schedule()deci nu este nevoie să le înregistrați singur.

Valorile sunt pentru fiecare execuție

Metricurile aparțin unei singure execuții a unei conducte. De fiecare dată când o conductă rulează, primește una nouă run_idiar valorile sale sunt stocate separat de cele ale executărilor anterioare. Într-o singură execuție, apel update_metric() din nou cu același nume suprascrie valoarea anterioară (ultima scriere câștigă).

Rularea programului a doua oară arată acest lucru: valorile din ambele rulări sunt reținute și distinse prin 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
                       
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

Acumulând cu increment = TRUE

Uneori, o măsurătoare este construită de-a lungul unei conducte, mai degrabă decât calculată o singură dată, cum ar fi numărarea apelurilor API într-o buclă sau numărul de încercări pentru o sarcină nesigură. Setare increment = TRUE adaugă value la valoarea curentă a valorii în loc să o suprascrieți. Dacă valoarea nu a fost setată încă în runda curentă, primul apel pur și simplu o setează. Un negativ value scade.

#' ./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)
      
      # 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 (the default) is fine for a one-off value
  update_metric("rows_read", nrow(df))
  
  df
}

În acest exemplu, ambele api_calls şi attempts acumula. Când se termină, api_calls este de 3 (unul pe pagină) în timp ce attempts este de cel puțin 3, în funcție de câte reîncercări au fost necesare. Fiecare valoare se termină cu o singură valoare pentru fiecare cursă de conductă.

Pentru detalii complete, consultați vigneta cu valori.

Alte modificări recente

Dacă ați ratat versiunile 1.3.x, există câteva lucruri mai mici care merită menționate:

• Noul @maestroCascadeTags eticheta propagă etichetele de metadate selectate (label, flagsși/sau loglevel) de la o conductă la toate conductele sale din aval într-un DAG. Fără valoare, le face pe toate trei în cascadă.
• get_slot_usage() câştigat min_datetime şi max_datetime parametri astfel încât fereastra de timp să poată fi fixată mai degrabă decât ancorată la ora curentă.

Lista completă a modificărilor este în notele de lansare.

Ca întotdeauna, orchestrare fericită!