Modern data engineering pipelines often have multiple transformation steps. Starting from reading the raw data, cleaning, filtering, validating and finally exporting to CSV or to some other format. R provides many powerful tools like dplyr for data manipulation. However, there is no native mechanism to visualize the state of a dataframe any given time as it moves through a pipeline.

Such a visualization is very useful since it helps warn if any of the transformation steps are going wrong for instance: whether a merge is creating unexpected duplicate columns or any row is silently being dropped. This is where DiagrammeR becomes extremely useful.

DiagrammeR allows us to create graphical pipeline diagrams that show:

• The sequence of operations
• The sate of the dataframe at each step
• Row counts, column counts, or other metadata
• Branching validation steps etc.

This article demonstrates how to use DiagrammeR along with a dynamic pipeline to create a data-aware visualisation for a real-world ETL workflow.

Why Visualize a Data Pipeline?

Visualizing a pipeline helps us to :

• Understand the flow of transformations
• Detect where rows are added or removed
• Document your ETL process for stakeholders
• Debug unexpected changes in row counts
• Communicate the logic behind your data preparation

And, especially for large datasets—like the one I am dealing with in the example, with 170+ million transactions (rows), this becomes especially valuable.

Tracking Pipeline State in R

DiagrammeR cannot directly inspect R objects. So the first step is to create a pipeline tracker that records the state of your dataframe after each transformation.

Pipeline Tracker

## R Code
pipeline <- list()
track <- function(name, df) {
  pipeline[[name]] <<- list(
    rows = nrow(df),
    cols = ncol(df),
    names = names(df)
  )
}

The Data Engineering Pipeline

# R Code
large_trx <- read_csv(file.path(input_dir, "archive/HI-Large_Trans.csv"))
track("read_csv", large_trx)
large_trx <- large_trx %>% rename(...)
track("rename_columns", large_trx)
large_trx <- large_trx %>% mutate(...)
track("mutate_bank_ids", large_trx)
large_trx <- large_trx %>%
  filter(!(sender_bank_id == receiver_bank_id &
           sender_account_id == receiver_account_id))
track("remove_self_transfers", large_trx)
write.csv(large_trx, file.path(output_dir, "transactions_hi_baselined.csv"))
track("write_csv", large_trx)

Building a DiagrammeR Pipeline Graph

DiagrammeR uses DOT language to define graphs. However, DOT syntax uses {} heavily, which conflicts with R’s glue interpolation.

# R Code
dot <- glue_data(
  .x = pipeline,
  .open = "<<",
  .close = ">>",
'
digraph pipeline {
  graph [layout = dot, rankdir = TB]
  start [shape = circle, label = "Start", fillcolor = "#A7C7E7"]
  read_csv        [shape = box, style = filled, fillcolor = lightblue,
                   label = "Read CSV\\nRows: <<read_csv$rows>>"]
  rename_cols     [shape = box, style = filled, fillcolor = lightblue,
                   label = "Rename Columns\\nRows: <<rename_columns$rows>>"]
  mutate_types    [shape = box, style = filled, fillcolor = lightblue,
                   label = "Convert Bank IDs\\nRows: <<mutate_bank_ids$rows>>"]
  filter_self     [shape = box, style = filled, fillcolor = lightblue,
                   label = "Remove Self-Transfers\\nRows: <<remove_self_transfers$rows>>"]
  write_csv       [shape = box, style = filled, fillcolor = lightblue,
                   label = "Write Cleaned File\\nRows: <<write_csv$rows>>"]
  end [shape = oval, label = "End", fillcolor = "#C6E0B4"]
  start -> read_csv
  read_csv -> rename_cols
  rename_cols -> mutate_types
  mutate_types -> filter_self
  filter_self -> write_csv
  write_csv -> end
}
')

And, finally render the diagram with

# R Code
grViz(dot)

What the Diagram Shows

The code generates the diagram shown on the right.

• A round start node
• A top-to-bottom pipeline
• Each transformation step
• Dynamic row counts pulled from your actual dataframe
• An oval end node

This gives us a living ETL diagram that updates automatically whenever the pipeline changes.

Obviously, the pipeline can be extended with schema validation, details of missing ids, internal vs. external transfers, amount inconsistency checks etc.