Skip to content

DataFrame Operations

Core row- and column-level transforms: filter, select, add or modify columns, sort, deduplicate, and the string/date/list expression namespaces. Every FlowFrame method mirrors its Polars counterpart and accepts an optional description that shows up as node documentation in the visual editor.

The flowfile_formula examples below use the Flowfile formula language; everything else is a Polars expression.

A worked example

This runs against committed data and is executed by the docs test suite:

import flowfile as ff

# Deduplicate, derive columns, filter, sort, then drop a column with a selector.
cleaned = (
    ff.read_csv("data/templates/orders.csv")
    .unique(subset=["order_id"])
    .with_columns(
        (ff.col("quantity") * ff.lit(2)).alias("double_qty"),
        ff.col("quantity").cast(ff.Float64).alias("quantity_f"),
    )
    .filter(ff.col("quantity") >= 3)
    .sort("quantity", descending=True)
    .select(ff.col("*").exclude("product_id"))
    .collect()
)

# String and date namespaces on expressions.
enriched = (
    ff.read_csv("data/templates/customers.csv")
    .with_columns(
        ff.col("name").str.to_uppercase().alias("name_upper"),
        ff.col("email").str.slice(0, 5).alias("email_prefix"),
        ff.col("name").str.contains("a").alias("name_has_a"),
        ff.col("signup_date").dt.year().alias("signup_year"),
        ff.col("signup_date").dt.weekday().alias("signup_weekday"),
    )
    .collect()
)

# A running total per customer with a window (cum_sum over a group).
running = (
    ff.read_csv("data/templates/orders.csv")
    .sort("order_date")
    .with_columns(ff.col("quantity").cum_sum().over("customer_id").alias("running_qty"))
    .collect()
)

The sections below break down each operation.

Filtering

import flowfile as ff

df = ff.FlowFrame({"price": [10, 20, 30], "qty": [5, 0, 10]})

# Polars expression predicate
df = df.filter(ff.col("price") > 15)

# With a description (surfaces in the visual editor)
df = df.filter(ff.col("price") > 15, description="Keep items over $15")

# Flowfile formula syntax
df = df.filter(flowfile_formula="[price] > 15 and [qty] > 0")

Which node the filter becomes

filter(flowfile_formula=...) emits an editable Filter node. A plain filter(ff.col(...) > x) predicate emits a polars_code node instead — the result is identical, but only the formula form is editable in the visual editor.

Selecting columns

# Select specific columns by name
df = df.select(["price", "qty"])

# Select with expressions
df = df.select([
    ff.col("price"),
    ff.col("qty").alias("quantity"),
])

# Keep everything except one column with a column selector.
# There is no ff.exclude() — use ff.col("*").exclude(...) or the selectors module.
df = df.select(ff.col("*").exclude("internal_id"))

Selectors

ff.numeric(), ff.string(), ff.all_(), and the other selector helpers pick columns by dtype or pattern — e.g. df.select(ff.numeric()) keeps only numeric columns.

Adding and modifying columns

# Expression form
df = df.with_columns([
    (ff.col("price") * ff.col("qty")).alias("total"),
])

# Flowfile formula form
df = df.with_columns(
    flowfile_formulas=["[price] * [qty]"],
    output_column_names=["total"],
    description="Calculate line totals",
)

Sorting

df = df.sort("price")
df = df.sort("price", descending=True)

# Multi-column sort
df = df.sort(["category", "price"], descending=[False, True])

Removing duplicates

# Drop fully duplicate rows
df = df.unique()

# Deduplicate on a subset of columns
df = df.unique(subset=["product_id"])

No drop_duplicates

FlowFrame does not expose drop_duplicates. Use unique() (optionally with subset=[...] and keep="first").

String operations

df = df.with_columns([
    ff.col("name").str.to_uppercase().alias("name_upper"),
    ff.col("code").str.slice(0, 3).alias("prefix"),
    ff.col("text").str.contains("pattern").alias("has_pattern"),
])

Conditional logic

df = df.with_columns([
    ff.when(ff.col("price") > 100)
    .then(ff.lit("Premium"))
    .when(ff.col("price") > 50)
    .then(ff.lit("Standard"))
    .otherwise(ff.lit("Budget"))
    .alias("tier"),
])

Date operations

df = df.with_columns([
    ff.col("date").dt.year().alias("year"),
    ff.col("date").dt.month().alias("month"),
    ff.col("date").dt.day().alias("day"),
    ff.col("date").dt.weekday().alias("weekday"),
])

Polars renames apply

The expression namespaces track the pinned Polars version: use dt.weekday() (not day_of_week) and cum_sum() (not cumsum).

List operations

df = df.with_columns([
    ff.col("tags").list.len().alias("tag_count"),
    ff.col("values").list.sum().alias("total"),
    ff.col("items").list.first().alias("first_item"),
])

Polars compatibility

Most Polars Expr methods are available. See the Polars docs for the full method reference; a few methods are renamed or fall back to polars_code nodes — see Expressions.


← Previous: Data Types | Next: Aggregations →