Chapter 9 Project 7: Automated Reporting Pipeline (ETL)

Manual reporting is error-prone, time-consuming, and demoralizing. In this chapter, we build a fully automated reporting pipeline using targets — R’s answer to Makefiles and workflow orchestrators like Airflow. The pipeline extracts data from multiple sources, transforms it, loads it to a data warehouse, and generates executive reports.

9.1 Business Context

Finance and executive teams need weekly reports covering:

• P&L statements by department and month
• Sales metrics including bookings, pipeline, and win rates
• Headcount analysis with cost-per-employee trends

The pipeline must be reproducible, scheduled, and failure-tolerant.

9.2 Architecture Overview

NetSuite ──┐
ADP ───────┼──► targets Pipeline ──► Snowflake ──► R Markdown Report ──► Email
Salesforce ─┘         (R)                (DB)           (PDF/HTML)        (blastula)

9.3 The targets Pipeline

targets tracks dependencies, skips up-to-date steps, and parallelizes execution.

# _targets.R
library(targets)
library(tarchetypes)
library(tidyverse)
library(DBI)
library(bigrquery)

tar_option_set(
  packages = c("tidyverse", "DBI", "bigrquery", "lubridate", "blastula"),
  format = "qs"  # Fast serialization with qs package
)

tar_source("R/")  # Load all functions from R/ directory

list(
  # ===== EXTRACT =====
  tar_target(netsuite_revenue, extract_netsuite_revenue()),
  tar_target(netsuite_expenses, extract_netsuite_expenses()),
  tar_target(adp_headcount, extract_adp_headcount()),
  tar_target(salesforce_bookings, extract_salesforce_bookings()),

  # ===== TRANSFORM =====
  tar_target(monthly_pnl, build_monthly_pnl(
    netsuite_revenue, netsuite_expenses, adp_headcount
  )),
  tar_target(sales_metrics, build_sales_metrics(salesforce_bookings)),

  # ===== LOAD =====
  tar_target(
    load_to_snowflake,
    load_snowflake(monthly_pnl, sales_metrics),
    cue = tar_cue(mode = "always")  # Always run this step
  ),

  # ===== REPORT =====
  tar_render(executive_report, "reports/executive_summary.Rmd")
)

Tip: tar_cue(mode = "always") forces a target to run even if dependencies haven’t changed. Use this for side effects like database loads or API calls.

9.4 Running the Pipeline

# Visualize the pipeline graph
targets::tar_visnetwork()

# Run the pipeline
targets::tar_make()

# Run with parallel workers
targets::tar_make_future(workers = 4)

# Read results
targets::tar_read(monthly_pnl)

9.5 dbt-Style Transformations

We implement analytics transformations in pure R, following dbt principles: modular, testable, and documented.

build_monthly_pnl <- function(revenue, expenses, headcount) {
  # Revenue by month and department
  rev_monthly <- revenue %>%
    mutate(month = floor_date(recognized_date, "month")) %>%
    group_by(month, department) %>%
    summarise(revenue = sum(revenue_amount, na.rm = TRUE), .groups = "drop")

  # Expenses by category
  exp_monthly <- expenses %>%
    mutate(month = floor_date(transaction_date, "month")) %>%
    group_by(month, department) %>%
    summarise(
      cogs = sum(amount * (account_type == "COGS"), na.rm = TRUE),
      opex = sum(amount * (account_type == "OpEx"), na.rm = TRUE),
      rd_spend = sum(amount * (account_type == "R&D"), na.rm = TRUE),
      .groups = "drop"
    )

  # Headcount metrics
  hc_monthly <- headcount %>%
    mutate(month = floor_date(date, "month")) %>%
    group_by(month, department) %>%
    summarise(
      avg_headcount = mean(headcount, na.rm = TRUE),
      avg_salary = mean(avg_salary, na.rm = TRUE),
      .groups = "drop"
    )

  # Combine and calculate derived metrics
  rev_monthly %>%
    left_join(exp_monthly, by = c("month", "department")) %>%
    left_join(hc_monthly, by = c("month", "department")) %>%
    mutate(
      net_income = revenue - cogs - opex - rd_spend,
      gross_margin_pct = (revenue - cogs) / revenue,
      opex_per_head = opex / avg_headcount
    )
}

9.6 Automated Email Distribution

library(blastula)

send_executive_summary <- function(report_path, recipients) {
  email <- compose_email(
    body = md(c(
      "# Weekly Executive Summary",
      "",
      paste("Report generated:", format(Sys.time(), "%B %d, %Y")),
      "",
      "Key highlights from this week's data:"
    )),
    footer = md("_Automated report from Data Science Team_")
  )

  email %>%
    add_attachment(file = report_path) %>%
    smtp_send(
      to = recipients,
      from = "analytics@company.com",
      subject = paste("Weekly Executive Summary —", format(Sys.Date(), "%b %d, %Y")),
      credentials = creds_envvar(
        user = Sys.getenv("SMTP_USER"),
        pass_envvar = "SMTP_PASS",
        host = "smtp.company.com",
        port = 587,
        use_ssl = TRUE
      )
    )
}

9.7 Error Handling & Notifications

safe_tar_make <- function() {
  tryCatch(
    targets::tar_make(),
    error = function(e) {
      blastula::smtp_send(
        email = compose_email(
          body = md(paste("Pipeline failed with error:", e$message))
        ),
        to = "data-team@company.com",
        from = "alerts@company.com",
        subject = "❌ Pipeline Failure Alert",
        credentials = creds_envvar(...)
      )
      stop(e)
    }
  )
}

9.8 Scheduling with cronR

library(cronR)

# Create a cron job to run every Monday at 6 AM
cron_rscript(
  rscript = "run_pipeline.R",
  schedule = "0 6 * * 1",
  log_file = "logs/pipeline.log"
)

9.9 Exercises

1. Add data quality checks using pointblank that fail the pipeline if null rates exceed thresholds.
2. Implement incremental loads: only process new or changed records since the last run.
3. Create a tar_target() that uploads results to S3 instead of Snowflake.
4. Build a Shiny app that displays pipeline status and allows manual re-runs.

---

In the next chapter, we’ll design a rigorous A/B testing framework with sequential testing and Bayesian analysis.