Example 7: PostgreSQL -> multi-target fan-out with Filter

Data Pipeline examples

Why this example

An events table in PostgreSQL captures every state change in an order-processing system. Three downstream surfaces care about subsets of the stream:

• Snowflake wants TXN_SETTLED events for BI.
• S3 wants TXN_WARNING events archived for cold storage.
• Ops on-call wants an email when a TXN_CRITICAL event lands.

The Branch node is the decision point that routes each row to the right downstream node.

Helix Beverages context: Helix's D2C e-commerce platform writes every state change to a PostgreSQL event store. Settled transactions feed Snowflake for the weekly leadership review, warnings are archived to S3, and critical events page ops on-call. This recipe is the routing core of named problem 5 in the Helix overview: critical events have to reach the right surface within minutes.

At a glance

• Connectors used: PostgreSQL, Snowflake, AWS S3, SMTP
• Nodes used: Start, Source, Branch, Transform, Load (x2), Notification
• Schedule: every 5 minutes
• Direction: One-way fan-out (PG -> Snowflake, S3, or SMTP)

Canvas

Start

Source: PG.events

Filter: SETTLED?

Transform

Load: Snowflake

Filter: WARNING?

Load: S3 (JSONL)

Reshape: subject + body

Notification: SMTP

Chained Filters give you a multi-way decision without inventing a router-of-routers. Each Filter has a KEPT port and a DROP port, and DROP flows into the next Filter, so nothing is thrown away on the way down the chain.

Step-by-step

#	Node	Action	Key configuration	Output shape
1	Start	None	None	None
2	Source	PostgreSQL: query	SELECT event_id, event_type, payload, occurred_at FROM events WHERE occurred_at > now() - interval '6 minutes'.	Event row
3	Filter	None	Rule: event_type equals TXN_SETTLED. KEPT -> Snowflake path. DROP -> downstream Filter.	Same row, routed
4	Transform (Snowflake path)	None	Map payload.amount and payload.currency to scalar columns.	Snowflake-shaped row
5	Load: Snowflake	Snowflake: Append	Target WH.FACT.TXN_SETTLED.	None
6	Filter (second)	None	Rule: event_type equals TXN_WARNING. KEPT -> S3. DROP (assumed CRITICAL) -> Notification path.	Same row, routed
7	Load: S3	S3: Upload File	Bucket events-cold, key warnings/dt=[occurred_at as yyyy-MM-dd]/[event_id].jsonl. Format JSONL. Append mode.	None
8	Reshape	None	Build subject = "Critical event " + row.event_id and body = JSON.stringify(row.payload, null, 2).	{ subject, body, recipients }
9	Notification	SMTP: Send Email	Recipients from a Variable (\{\{var.global.opsOncallEmail\}\}). Subject + body from the Reshape output.	None

Variations

• More than three targets: chain another Filter (or use a Router node, which has five route ports plus a default) for an N-way decision.
• Different DB family: swap the Source for MSSQL / MySQL / Oracle / Snowflake. The Filter rules stay identical.
• Send all three event types to S3 too: add a Fork right after Source, then run the existing Filter chain on one Fork output; the other Fork output goes to a single Load: S3 with all events.
• Replace Notification with PagerDuty webhook: swap SMTP for the Generic API connector pointed at the PagerDuty Events v2 endpoint.