You Don’t Need Airflow: Orchestrate Many Data Flows in R with Maestro

Pipelines

I’ve created three pipelines that each retrieve data from one of the eBird endpoints and stores it in a duckdb table. Each pipeline is scheduled to run at a particular time interval so that new data is regularly inserted into the table.

#' @maestroFrequency 3 hours
#' @maestroStartTime 2025-02-20 12:00:00
#' @maestroTz America/Halifax
get_nearby_notable_obs <- function() {

  req <- httr2::request("https://api.ebird.org/v2") |>
    httr2::req_url_path_append("data/obs/geo/recent/notable") |>
    httr2::req_url_query(
      lat = 44.88,
      lng = -63.52
    ) |>
    httr2::req_headers(
      `X-eBirdApiToken` = Sys.getenv("EBIRD_API_KEY")
    )

  resp <- req |>
    httr2::req_perform()

  obs <- resp |>
    httr2::resp_body_json(simplifyVector = TRUE) |>
    dplyr::mutate(
      insert_time = Sys.time()
    )

  # Connect to a local in-memory duckdb
  conn <- DBI::dbConnect(duckdb::duckdb())
  on.exit(DBI::dbDisconnect(conn))

  # Create and write to a table
  DBI::dbWriteTable(
    conn,
    name = "recent_notable_observations",
    value = obs,
    append = TRUE
  )
}

#' @maestroFrequency 1 day
#' @maestroStartTime 2025-02-20 18:00:00
#' @maestroTz America/Halifax
get_region_stats <- function() {

  now <- Sys.time()
  cur_year <- lubridate::year(now)
  cur_month <- lubridate::month(now)
  cur_day <- lubridate::day(now)

  req <- httr2::request("https://api.ebird.org/v2") |>
    httr2::req_url_path_append("product/stats", "CA-NS", cur_year, cur_month, cur_day) |>
    httr2::req_headers(
      `X-eBirdApiToken` = Sys.getenv("EBIRD_API_KEY")
    )

  resp <- req |>
    httr2::req_perform()

  stats <- resp |>
    httr2::resp_body_json(simplifyVector = TRUE) |>
    dplyr::as_tibble()

  # Connect to a local in-memory duckdb
  conn <- DBI::dbConnect(duckdb::duckdb())
  on.exit(DBI::dbDisconnect(conn))

  # Create and write to a table
  DBI::dbWriteTable(
    conn,
    name = "region_stats",
    value = stats,
    append = TRUE
  )
}

#' @maestroFrequency 1 day
#' @maestroStartTime 2025-02-20 15:00:00
#' @maestroTz America/Halifax
get_species_list <- function() {

  req <- httr2::request("https://api.ebird.org/v2") |>
    httr2::req_url_path_append("product/spplist", "CA-NS") |>
    httr2::req_headers(
      `X-eBirdApiToken` = Sys.getenv("EBIRD_API_KEY")
    )

  resp <- req |>
    httr2::req_perform()

  spec_list <- resp |>
    httr2::resp_body_json(simplifyVector = TRUE)

  spec_df <- dplyr::tibble(
    speciesCode = spec_list
  ) |>
    dplyr::mutate(
      insert_time = Sys.time()
    )

  # Connect to a local in-memory duckdb
  conn <- DBI::dbConnect(duckdb::duckdb())
  on.exit(DBI::dbDisconnect(conn))

  # Create and write to a table
  DBI::dbWriteTable(
    conn,
    name = "species_list",
    value = spec_df,
    append = TRUE
  )
}

Orchestrator

With the pipelines created we move to the orchestrator script. This is an R script or Quarto document that runs maestro functions to create the schedule from the tags and the run the schedule according to some frequency – a frequency that should always be at least as frequent as your most frequent pipeline.

library(maestro)

schedule <- build_schedule()

ℹ 3 scripts successfully parsed

run_schedule(
  schedule, 
  orch_frequency = "1 hour",
  check_datetime = as.POSIXct("2025-02-26 15:00:00", tz = "America/Halifax") # for reproducibility - in practice use Sys.time()
)

── [2025-02-26 15:14:23]
Running pipelines ▶ 
ℹ get_nearby_notable_obs
✔ get_nearby_notable_obs [758ms]

ℹ get_species_list
✔ get_species_list [108ms]


── [2025-02-26 15:14:24]
Pipeline execution completed ■ | 0.885 sec elapsed 
✔ 2 successes | → 1 skipped | ! 0 warnings | ✖ 0 errors | ◼ 3 total
────────────────────────────────────────────────────────────────────────────────

── Next scheduled pipelines ❯ 
Pipe name | Next scheduled run
• get_nearby_notable_obs | 2025-02-26 22:00:00
• get_region_stats | 2025-02-26 22:00:00
• get_species_list | 2025-02-27 19:00:00

── Maestro Schedule with 3 pipelines:  
• Success

status <- get_status(schedule)
status

# A tibble: 3 × 10
  pipe_name  script_path invoked success pipeline_started    pipeline_ended     
                                                
1 get_nearb… ./pipeline… TRUE    TRUE    2025-02-26 19:14:23 2025-02-26 19:14:24
2 get_regio… ./pipeline… FALSE   FALSE   NA                  NA                 
3 get_speci… ./pipeline… TRUE    TRUE    2025-02-26 19:14:24 2025-02-26 19:14:24
# ℹ 4 more variables: errors , warnings , messages ,
#   next_run 

We can run all this interactively, but the power of maestro is in running it scheduled in production. This way, the data will grow and update regularly. Deployment is not special in the case of maestro – you just need to be sure that the orchestrator is scheduled to run at the same frequency as specified in orch_frequency. Check out my previous post for a walk through of deployment on Google Cloud.

ChatGPT Conversation

The conversation below was with ChatGPT 4-o on February 26th 2025.

Me: I want to do some cost estimates for cloud computing. I have three proposed tech stacks, all of which I could run on any cloud provider. I don’t want to compare cloud providers but instead compare the estimated cost of each architecture. In all cases, the project would involve taking 10 data flows that process around 500MB-2GB of data at varying time intervals with the most frequent being every hour and the least being every day. These data flows could be Python or R scripts or something else depending on the tech stack.

Here are the three proposed tech stacks:

1. All 10 data flows run on a single instance and an orchestrator script runs every hour to determine which script to run. Importantly, all of this runs on a single serverless instance.
2. Each 10 data flow has its own serverless instance and a scheduling service from the cloud provider is used to configure each script to run at its required time interval.
3. An open-source Apache Airflow project is hosted on a compute instance.
4. An orchestration platform provided by the cloud provider is used out of the box.

For each scenario, provide some estimate as to the hardware requirements (e.g., instance size, compute nodes, etc.) and give a rough estimate of monthly cost in CAD.

I’ll break down the estimated cloud costs for each of the four approaches based on typical cloud pricing. Since we’re not comparing cloud providers directly, I’ll use common pricing models across AWS, GCP, and Azure. Prices will be in CAD.

---

Approach 1: Single Serverless Instance for All Data Flows

---

Approach 2: Separate Serverless Instance for Each Data Flow

---

Approach 3: Apache Airflow on a Compute Instance

---

Approach 4: Cloud Provider’s Managed Orchestration Service

---

Summary of Estimated Monthly Costs (CAD)

1. Single Serverless Instance	$100–120	Simple but limited scalability.
2. Separate Serverless Instances	$110–130	More parallelism but increased function calls.
3. Apache Airflow on Compute	$170–200	More control, but higher infra cost.
4. Cloud Managed Orchestration	$80–100	Scalable and cost-efficient.

---

Recommendation

• If cost is the main factor → Approach 4 (Managed Orchestration) is best.

• If you prefer full control → Approach 3 (Airflow).

• If you need balance between control and scalability → Approach 2 (Separate Serverless Instances).

Me: In the first scenario are you properly accounting for the fact that the job would only run once every hour? Even if there are multiple processes being executed, it is still one job scheduled to run hourly.

You’re absolutely right! In Scenario 1 (Single Serverless Instance Running an Orchestrator Script Hourly), the orchestrator script only runs once per hour, making the cost lower than I initially estimated. Let’s correct that calculation.

---

Scenario 1: Revised Cost Calculation

• Execution Frequency:

  • The orchestrator runs once every hour → 24 runs per day.

  • Each run executes one or more data flows sequentially.

  • Assume an average runtime of 10 minutes per execution (to process multiple flows).

• Compute Usage:

  • Suppose the instance scales between 2–4 vCPUs and 4–8GB RAM, depending on the workload.

  • At peak, assume 2 vCPUs and 4GB RAM for the entire execution.

  • Cost estimation:

    • 24 runs/day × 30 days = 720 runs/month.

    • 10 minutes per run = 7,200 minutes per month = 120 compute hours/month.

    • Compute cost: $0.06 per vCPU-hour × 2 vCPUs × 120 hours = $14.40/month.

    • Memory cost: $0.009 per GB-hour × 4GB × 120 hours = $4.32/month.

• Storage & Data Transfer:

Total Estimated Cost: $25–35 CAD/month

---

Updated Cost Comparison (CAD)

1. Single Serverless Instance	$25–35 ✅	Most cost-effective but limited parallelism.
2. Separate Serverless Instances	$110–130	More parallelism but higher function call costs.
3. Apache Airflow on Compute	$170–200	Full control but highest infra cost.
4. Cloud Managed Orchestration	$80–100	Scalable and cost-efficient.

---

Final Thoughts

• Scenario 1 is now clearly the cheapest option (~$25–35 CAD/month).

• However, it lacks parallelism—if multiple heavy processes need to run, the single instance might take longer to complete.

• If cost is the priority, Scenario 1 is the best choice.

• If scalability and efficiency matter, Scenario 4 (Managed Orchestration) is a good balance.

Would you like a cost efficiency vs. performance trade-off chart to visualize these differences? 🚀

Me: Finally, let’s say for Scenario 1 that we enable parallel processing if needed. For example, if 4 processes need to run simultaneously it can scale up to 4 cores.

Great! If we allow Scenario 1 to scale dynamically to support parallel processing, we need to adjust the compute cost based on variable CPU utilization. Here’s how that impacts the estimation.

---

Scenario 1: Revised Cost with Parallel Processing

Assumptions:

• Orchestrator runs every hour (24 times per day).

• Each execution lasts 10 minutes but now can use up to 4 vCPUs when needed.

• Average vCPU usage per run:

  • Some runs may use 1–2 vCPUs (light workloads).

  • Others may spike to 4 vCPUs (heavy parallel processing).

  • Assume an average usage of 3 vCPUs per execution.

• Memory scales accordingly: Assume 6GB RAM on average.

Compute Cost Calculation:

• Total monthly runtime:

  • 10 min per execution × 24 runs/day × 30 days = 7,200 minutes (~120 hours/month).

• CPU Cost:

• Memory Cost:

• Storage & Data Transfer:

  • $5–10/month (no significant change).

Final Total Cost: ~$35–45 CAD/month

---

Final Cost Comparison (CAD)

1. Single Serverless Instance (Parallel Processing)	$35–45 ✅	Cheapest, allows parallelism but limited by single-instance performance.
2. Separate Serverless Instances	$110–130	Fully scalable, better for frequent workloads.
3. Apache Airflow on Compute	$170–200	Self-managed, full control, higher maintenance.
4. Cloud Managed Orchestration	$80–100	Best for scalability without management overhead.

---

Conclusion

• Scenario 1 with parallelism remains the cheapest (~$35–45 CAD/month).

• It allows parallel execution when needed, reducing the risk of long-running sequential jobs.

• It’s best if workload concurrency is manageable (e.g., not exceeding 4 vCPUs often).

• If greater scalability is required, Scenario 4 (Managed Orchestration) would be the next best option.

Would you like a cost vs. scalability chart to visualize these trade-offs? 🚀