Joins

Combine two FlowFrames on one or more keys, stack frames vertically, and validate keys before joining. Joins support the strategies inner, left, right, full, semi, anti, outer, plus cross.

A worked example

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

import flowfile as ff

orders = ff.read_csv("data/templates/orders.csv")
customers = ff.read_csv("data/templates/customers.csv")
products = ff.read_csv("data/templates/products.csv")

# Enrich orders with customer and product attributes.
inner = orders.join(customers, on="customer_id", how="inner")
with_product = orders.join(products, on="product_id", how="left")

# full keeps unmatched rows from both sides; semi/anti filter the left side.
full = orders.join(customers, on="customer_id", how="full")
repeat_customers = customers.join(orders, on="customer_id", how="semi")
dormant_customers = customers.join(orders, on="customer_id", how="anti")

# Stack frames vertically with ff.concat (there is no vstack), then dedupe.
deduped = ff.concat([customers, customers]).unique()

# Validate a join key before joining: compare distinct keys to total rows.
key_stats = customers.select(
    ff.col("customer_id").n_unique().alias("distinct_customers"),
    ff.col("customer_id").count().alias("total_rows"),
)

inner_df = inner.collect()
full_df = full.collect()
repeat_df = repeat_customers.collect()
dormant_df = dormant_customers.collect()
key_stats_df = key_stats.collect()

The sections below break down each pattern.

Basic join

import flowfile as ff

customers = ff.FlowFrame({
    "id": [1, 2, 3],
    "name": ["Alice", "Bob", "Charlie"],
})

orders = ff.FlowFrame({
    "order_id": [101, 102, 103],
    "customer_id": [1, 2, 1],
    "amount": [100, 200, 150],
})

result = customers.join(
    orders,
    left_on="id",
    right_on="customer_id",
    how="inner",
    description="Join customers with orders",
)

Join types

# Inner join (default): rows with a match on both sides
df1.join(df2, on="key", how="inner")

# Left join: all left rows, matched right columns or null
df1.join(df2, on="key", how="left")

# Right join: all right rows
df1.join(df2, on="key", how="right")

# Full join: all rows from both sides (unmatched filled with null)
df1.join(df2, on="key", how="full")

# Semi join: left rows that have a match (keeps only left columns)
df1.join(df2, on="key", how="semi")

# Anti join: left rows with no match
df1.join(df2, on="key", how="anti")

full, not outer

Under the pinned Polars version the full outer strategy is how="full". how="outer" is accepted as a legacy alias, but new code should use full.

Multiple join keys

result = df1.join(
    df2,
    on=["region", "year"],  # join on several columns
    how="inner",
)

# Different column names on each side
result = df1.join(
    df2,
    left_on=["region_code", "period"],
    right_on=["region", "year"],
    how="left",
)

Cross join

# Cartesian product of both frames
result = df1.join(df2, how="cross")

Stacking frames

There is no vstack on FlowFrame — use ff.concat to stack frames vertically:

# Vertical concatenation
combined = ff.concat([df1, df2, df3])

# Concatenate then drop duplicate rows
union_df = ff.concat([df1, df2]).unique()

# Diagonal concatenation aligns differing schemas
combined = ff.concat([df1, df2], how="diagonal")

Join validation

FlowFrame has no frame-level len() or n_unique(). Count distinct keys with an expression inside select, and count matched rows by collecting a small aggregate rather than taking len() of a frame:

# Are there duplicate keys in the right table?
key_stats = df2.select(
    ff.col("customer_id").n_unique().alias("distinct_keys"),
    ff.col("customer_id").count().alias("total_rows"),
).collect()
if key_stats["distinct_keys"][0] < key_stats["total_rows"][0]:
    print("Warning: duplicate keys in right table")

# How many left rows found no match?
result = df1.join(df2, on="id", how="left")
unmatched = result.filter(ff.col("amount").is_null()).select(
    ff.col("id").count().alias("unmatched")
).collect()
print(f"Unmatched records: {unmatched['unmatched'][0]}")

join_asof and join_where are not supported

These methods are present on FlowFrame (they are injected from Polars) but raise at call time — Flowfile has no native node for them. Use raw Polars in a Python Script node for time- or predicate-based joins.

---

← Previous: Aggregations | Next: Cloud Storage →