Collecting and filtering output"
In pipeflow: Fast Interactive Data Analysis Pipelines

knitr::opts_chunk$set(
    comment = "#",
    prompt = FALSE,
    tidy = FALSE,
    cache = FALSE,
    collapse = TRUE
)

old <- options(width = 100L)

Generally speaking, one should keep pipeline steps as simple as possible, basically following the principle "one step, one task". Splitting up analysis steps into multiple functions naturally can be hard to manage, but since {pipeflow} manages all function and parameter dependencies for you, this is not a problem.

Following this principle, usually a lot of pipeline steps will carry intermediate results and only a few steps will contain the final output we are interested in. This vignette shows how to conveniently tag, collect, and filter pipeline outputs using tags and views.

Tagging steps

Tags let you label steps with meaningful keywords so you can later filter the pipeline by topic, stage, or output type. Tags can be set during pipeline creation or later using pip_set_tags().

library(pipeflow)

pip <- pip_new("my-pip") |>
    pip_add(
        "data",
        function(data = airquality) data
    ) |>
    pip_add("data_prep",
        function(data = ~data) {
            replace(data, "Temp.Celsius", (data[, "Temp"] - 32) * 5 / 9)
        },
        tags = "data"                   # <-- set 'data' tag
    ) |>
    pip_add(
        "data_summary",
        function(
            data = ~data_prep,
            xVar = "Temp.Celsius",
            yVar = "Ozone"
        ) {
            format(summary(data[, c(xVar, yVar)])) |>
                as.data.frame(row.names = NA)
        },
        tags = c("data", "summary")     # <-- set 'data' and 'summary' tags
    ) |>
    pip_add(
        "data_plot",
        function(
            data = ~data_prep,
            xVar = "Temp.Celsius",
            yVar = "Ozone"
        ) {
            require(ggplot2, quietly = TRUE)
            ggplot(data) +
                geom_point(aes(.data[[xVar]], .data[[yVar]])) +
                labs(title = "Data")
        },
        tags = c("data", "plot")        # <-- set 'data' and 'plot' tags
    ) |>
    pip_add(
        "model_fit",
        function(
            data = ~data_prep,
            xVar = "Temp.Celsius",
            yVar = "Ozone"
        ) {
            lm(paste(yVar, "~", xVar), data = data)
        },
        tags = c("model", "fit")        # <-- set 'model' and 'fit' tags
    ) |>
    pip_add(
        "model_summary",
        function(fit = ~model_fit) {
            summary(fit) |>
                coefficients() |>
                as.data.frame()
        },
        tags = c("model", "summary")    # <-- set 'model' and 'summary' tags
    ) |>
    pip_add(
        "model_plot",
        function(
            model = ~model_fit,
            data_plot = ~data_plot,
            xVar = "Temp.Celsius",
            yVar = "Ozone"
        ) {
            coeffs <- coefficients(model)
            data_plot +
                geom_abline(intercept = coeffs[1], slope = coeffs[2]) +
                labs(title = "Linear model fit")
        },
        tags = c("model", "plot")       # <-- set 'model' and 'plot' tags
    )

We used two families of tags: "data"/"model" to distinguish the topic, and "summary"/"plot"/"fit" for the output type. Whenever tags are defined, they are shown in the pipeline overview:

pip

Before showing how to make use of the tags, let's run the pipeline and inspect the output individually as we did in the previous vignettes.

pip_run(pip)

pip

pip[["data_plot", "out"]]

pip[["model_plot", "out"]]

Flat output collection

pip_collect_out() returns all step outputs as a flat named list.

out <- pip_collect_out(pip)

names(out)

Filtered output using tags

To collect only the output of steps with a specific tag, we use pip_view(), which is {pipeflow}'s general-purpose function for filtering pipelines, and then call pip_collect_out() on the filtered pipeline. To collect only the plots, for example, we can filter by the plot tag.

pip_view(pip, tags = "plot")

pip |>
    pip_view(tags = "plot") |>
    pip_collect_out() |>
    gridExtra::grid.arrange(grobs = _, nrow = 2)

Grouped output via views

Grouped output is achieved by composing pip_view() with pip_collect_out(). For example, to collect outputs grouped by topic ("data" and "model"):

grouped <- list(
    data  = pip_view(pip, tags = "data")  |> pip_collect_out(),
    model = pip_view(pip, tags = "model") |> pip_collect_out()
)

names(grouped[["data"]])
names(grouped[["model"]])

More on views

To make the example a bit more interesting, we first update some parameters.

pip_set_params(pip, params = list(xVar = "Solar.R", yVar = "Wind"))

pip

{pipeflow} views provide a variety of filtering options. In the previous section, the filtering was done based on tags, but you can also filter based on other properties, for example, all steps that depend on the model_fit step:

pip |> pip_view(filter = list(depends = "model_fit", state = "outdated"))

or using regex-based filtering, for example, to filter all outdated steps starting with data:

pip |>
    pip_view(filter = list(step = "^data", state = "outdated"), fixed = FALSE)

Views can also be chained together:

v <- pip |> pip_view(filter = list(state = "outdated"))
v

v2 <- v |> pip_view(tags = "plot")
v2

Last but not least, views can be run as pipelines themselves, which allows to conveniently re-run only the filtered steps, while {pipeflow} ensures that any upstream dependencies are run first if needed.

v2 |> pip_run()

Having a closer look at the run log, you'll see which steps were re-run as part of the [view] and which were re-run as [upstream] dependencies. Since all views work by reference on the given pipeline, the original pipeline is now up-to-date for the filtered steps.