Python and R Blocks

Python and R blocks let you perform custom analysis inside a Data Pipeline.

The FlaskTrack pipeline runtime gives your script access to the inputs connected to the block and lets the script create declared outputs.

Python blocks

Start Python scripts with:

import flasktrack as ft

Read a table

table = ft.input_table("input")

For Pandas:

df = ft.input_table("input").to_pandas()

Example:

import flasktrack as ft

df = ft.input_table("input").to_pandas()

df["normalized"] = (
    df["value"] / df["value"].mean()
)

ft.output_table("output", df)

Read JSON

metadata = ft.input_json("metadata")

Example:

threshold = metadata["threshold"]

Read a file

path = ft.input_file("raw_file")

Use the returned path with the library appropriate for the file type.

Example:

import pandas as pd
import flasktrack as ft

path = ft.input_file("raw_file")
df = pd.read_csv(path)

ft.output_table("output", df)

Python outputs

Arrow table

ft.output_table("output", df)

JSON

ft.output_json(
    "summary",
    {
        "rows": len(df),
        "mean": float(df["value"].mean()),
    },
)

Parquet

ft.output_parquet(
    "result",
    df,
    compression="snappy",
)

File

If your script creates a file:

df.to_csv(
    "/workspace/output/result.csv",
    index=False,
)

ft.output_file(
    "result",
    "/workspace/output/result.csv",
    content_type="text/csv",
)

Python logging

Add messages to the run history with:

ft.log("Starting analysis")

You can also use levels:

ft.log("Removed 4 invalid rows", level="warning")

Available levels are:

debug
info
warning
error

R blocks

R blocks use the FlaskTrack R helpers.

Start with:

source("/opt/flasktrack/flasktrack.R")

Read a table

data <- ft_input_table("input")

Example:

source("/opt/flasktrack/flasktrack.R")

data <- ft_input_table("input")

data$normalized <- (
    data$value / mean(data$value)
)

ft_output_table("output", data)

Read JSON

metadata <- ft_input_json("metadata")

Read a file

path <- ft_input_file("raw_file")

R outputs

Arrow table

ft_output_table("output", data)

JSON

ft_output_json(
    "summary",
    list(
        rows = nrow(data),
        mean = mean(data$value)
    )
)

Parquet

ft_output_parquet(
    "result",
    data,
    compression = "snappy"
)

File

write.csv(
    data,
    "/workspace/output/result.csv",
    row.names = FALSE
)

ft_output_file(
    "result",
    "/workspace/output/result.csv",
    "text/csv"
)

R logging

ft_log("Starting analysis")

With a level:

ft_log("Missing values were removed", "warning")

Matching script names to block ports

If your block input is named:

measurements

use:

ft.input_table("measurements")

or:

ft_input_table("measurements")

If your output is named:

summary

use:

ft.output_json("summary", value)

The script cannot create arbitrary undeclared pipeline outputs.

Multiple inputs

A Python or R block can combine several inputs.

Example graph:

Measurements ──► measurements
                   │
                   │
Sample metadata ─► Python
                   │
                   ▼
                output

Python:

import flasktrack as ft

measurements = ft.input_table("measurements").to_pandas()
sample_info = ft.input_table("sample_info").to_pandas()

merged = measurements.merge(
    sample_info,
    on="sample_id",
)

ft.output_table("output", merged)

Runtime limits

Each Python or R block has execution limits such as:

timeout;
memory;
maximum generated output size.

The defaults are suitable for typical transformations.

Increase them only when the workload actually requires more resources.

Packages and internet access

Pipeline scripts run without general network access.

Do not rely on installing packages while the pipeline is running.

If your organization needs an additional Python or R library, it must be included in the configured pipeline runtime by the FlaskTrack administrator.

This helps keep repeated pipeline runs consistent.