Turning a pipeline into Python

An exported pipeline file says exactly what every node does. This page tells a person, or an AI assistant, how to write it as PySpark or pandas code.

Using an AI assistant

Export the pipeline, then give the file to your AI assistant with this request. An assistant that can read web pages fetches this guide itself; otherwise paste the guide's text file, https://www.trinium.io/pipeline-to-python.txt, along with the file.

I am giving you a Trinium pipeline file (JSON). First read
https://www.trinium.io/pipeline-to-python.txt, which explains the file and every node.
Then write this pipeline as one Python script using PySpark (or: pandas).
- One function per node, named after the node, called in the order the wires allow.
- Use the file's connection and file names as placeholders I will fill in.
- Follow the guide's rules on names, NULLs, integer division, row order and dates.
- Where the code cannot do exactly what a node does, say so in a comment there.

<paste the pipeline file here>

Treat the answer as a draft: run it on the same data as the pipeline and compare the results before you rely on it (see Checking the result).

The file

A pipeline file is one JSON object. Trinium writes it; import reads files Trinium wrote, so do not write one by hand.

A node has an id, a node_type (the headings below), a label (its name on the canvas), its config (its settings) and a position. Nodes of type documentation and container are notes and boxes drawn on the canvas: they do nothing, so skip them.

An edge is a wire: from_node and from_port to to_node and to_port. A port is a named anchor, such as a Filter's true and false; each node below lists its ports. Several wires may feed one input port: their rows are combined. An edge from a port named files carries a list of files, not rows, to a File Action.

A connection is named by its ref in a node's connection_id. The file never holds a connection's address or credentials: put your own in their place.

The rules that change answers

The same code can give a different answer in Trinium and in Python. Keep these in your code:

• Order. Run each node after the nodes wired into it. Every node can start as soon as its inputs have rows; only the result matters.
• Column names are case-exact. Amount and amount are two columns. Spark ignores case unless you set spark.sql.caseSensitive to true; pandas is case-exact already.
• Expressions are SQL. Filter, Formula, Route and the other nodes that take a rule hold a SQL expression. [Order Date] in square brackets names a column, as does "Order Date" in double quotes; single quotes are text. In Spark write the column in backticks inside F.expr("...") or spark.sql; in pandas use df["Order Date"].
• Integer division stays an integer. 7 / 2 is 3 in Trinium when both sides are integers. Spark returns 3.5 (use div or cast); in pandas use //.
• NULL is not a value. A comparison with NULL is not true, so a Filter sends such a row to false. A join never matches a NULL key. pandas treats missing keys as equal when merging, so drop or fill them first.
• Dates are ISO. A date is YYYY-MM-DD and a date-time YYYY-MM-DD HH:MM:SS. A date in an expression is written DATE '2026-01-31'.
• Row order matters for a few nodes. Unique keeps the first row it saw, Record ID numbers rows as they arrive, Sample takes the first rows, and Multi-Row Formula looks at the previous row. Spark has no row order across partitions: number the rows where you read them, and sort by that number where order matters.
• Text comparisons in Filter and Route ignore case in basic mode unless case_sensitive is true.
• A failed type conversion is NULL, not an error, in Manage Columns. Spark 4 (Databricks Runtime 17 and later) raises an error on a bad cast unless you use try_cast.

Parameters and date tokens

A setting can hold ${name}, replaced by the parameter's value when the pipeline runs (its default when none is given). These built-in tokens work in any setting:

In a REST node's path, body and headers, ${column} is instead the current row's value of that column.

Expressions in Python

Most SQL in a pipeline file runs unchanged in Spark through F.expr or spark.sql. These functions differ:

Regular expressions in a pipeline have no look-around and no back-references, so they work unchanged in Python's re and in Spark; only a replacement's group reference changes (\1 in Python, $1 in Spark).

What has no Python equivalent

These belong to the platform that runs the code, not to the pipeline's logic:

• Bookmarks: reading only new files or new rows since the last successful run. Keep that list or high-water mark yourself.
• Schedules, file-arrival triggers, Flows and alerts: your scheduler's job.
• Staging and Production: run the code against each environment's connections.
• Test runs: a Trinium Test run reads 50 rows and writes nothing.

Every node

Each node lists its ports (the names its wires use), the settings in its config, and how to write it in PySpark and in pandas. In the code, df is the node's input, F is pyspark.sql.functions and W is pyspark.sql.Window.

Input

Apache Iceberg input: iceberg_input

Reads an Apache Iceberg table through its catalog (REST, AWS Glue or S3 Tables), optionally at a past snapshot.

Ports. In: none (a source). Out: output.

PySpark

df = spark.read.format("iceberg").option("snapshot-id", sid).load(f"{catalog}.{namespace}.{table}")

pandas

df = pyiceberg.catalog.load_catalog("c", **props).load_table(f"{namespace}.{table}").scan().to_pandas()

Athena: athena_input

Reads a table, or a query's result, from Amazon Athena.

Ports. In: none (a source). Out: output.

PySpark

df = spark.table(f"{database}.{table}")  # in Glue, the same table through the Glue catalog

pandas

df = awswrangler.athena.read_sql_query(query, database=db)

Azure Blob Files: azure_blob_input

Reads one blob, or every matching blob under a folder, from Azure Blob Storage.

Ports. In: none (a source). Out: output, files.

PySpark

df = spark.read.parquet(f"abfss://{container}@{account}.dfs.core.windows.net/{prefix}")

pandas

df = pd.read_parquet(f"abfs://{container}/{prefix}", storage_options={...})  # needs adlfs

BigQuery input: bigquery_input

Reads a table, or a query's result, from Google BigQuery.

Ports. In: none (a source). Out: output.

PySpark

df = spark.read.format("bigquery").option("query", query).option("viewsEnabled", True).load()

pandas

df = bigquery.Client(project=p).query(query).to_dataframe()

ClickHouse input: clickhouse_input

Reads a table, or a query's result, from ClickHouse.

Ports. In: none (a source). Out: output.

PySpark

df = spark.read.format("jdbc").option("url", "jdbc:clickhouse://host:8123/db").option("query", query).load()

pandas

df = clickhouse_connect.get_client(**creds).query_df(query)

CSV Files: csv_input

Reads CSV files uploaded to the pipeline's file library, as one table.

Ports. In: none (a source). Out: output.

PySpark

df = spark.read.csv(paths, header=True, inferSchema=True, sep=",")

pandas

df = pd.concat([pd.read_csv(p) for p in paths], ignore_index=True)

Trinium matches the columns of later files to the first file that can be read, BY NAME: a missing column is NULL, an extra one is dropped, a file sharing no column, or one that cannot be read, is skipped. Spark's CSV reader matches by POSITION, so read each file and align by name (unionByName with allowMissingColumns=True) to keep that rule.

Databricks input: databricks_input

Reads a table, or a query's result, from a Databricks SQL warehouse.

Ports. In: none (a source). Out: output.

PySpark

df = spark.sql(query)  # inside Databricks

pandas

with databricks.sql.connect(**creds) as c: df = pd.read_sql(query, c)

Excel Files: excel_input

Reads sheets from uploaded Excel workbooks.

Ports. In: none (a source). Out: output.

PySpark

pdf = pd.read_excel(path, sheet_name=sheet, skiprows=skip_rows); df = spark.createDataFrame(pdf)

pandas

df = pd.read_excel(path, sheet_name=sheet_name or 0, skiprows=skip_rows)  # sheet_name=None reads all

File from URL: url_input

Downloads one file from a web link and reads it.

Ports. In: none (a source). Out: output, files.

PySpark

pdf = pd.read_csv(url); df = spark.createDataFrame(pdf)  # or download with requests first

pandas

df = pd.read_csv(io.BytesIO(requests.get(url, timeout=60).content))

GCS Files: gcs_input

Reads one object, or every matching object under a folder, from Google Cloud Storage.

Ports. In: none (a source). Out: output, files.

PySpark

df = spark.read.parquet(f"gs://{bucket}/{prefix}")

pandas

df = pd.read_parquet(f"gs://{bucket}/{prefix}")  # needs gcsfs

Google Sheets input: google_sheets_input

Reads a tab of a Google Sheets spreadsheet.

Ports. In: none (a source). Out: output.

PySpark

pdf = pd.DataFrame(gspread.service_account().open_by_key(sid).worksheet(tab).get_all_records()); df = spark.createDataFrame(pdf)

pandas

df = pd.DataFrame(gspread.service_account().open_by_key(sid).worksheet(tab).get_all_records())

A code with a leading zero, such as 007, stays text.

HubSpot: hubspot_input

Reads HubSpot CRM objects (contacts, companies, deals and so on).

Ports. In: none (a source). Out: output.

PySpark

# page through GET /crm/v3/objects/{object}?properties=...&after=... with requests, then spark.createDataFrame

pandas

df = pd.json_normalize(results)  # results gathered from /crm/v3/objects/{object}

Jira: jira_input

Reads Jira issues with a JQL query.

Ports. In: none (a source). Out: output.

PySpark

# page through /rest/api/3/search?jql=...&startAt=... with requests, then spark.createDataFrame

pandas

df = pd.json_normalize(issues)  # issues gathered from /rest/api/3/search

JSON Files: json_input

Reads JSON files (an array of objects, or one object per line) uploaded to the pipeline's file library.

Ports. In: none (a source). Out: output.

PySpark

df = spark.read.json(paths, multiLine=True)  # multiLine=False for one object per line

pandas

df = pd.concat([pd.read_json(p) for p in paths], ignore_index=True)  # lines=True for NDJSON

List Files: list_files

Lists the files in a folder, one row per file, without reading them: path, name, extension, parent, size_bytes, modified, change_token.

Ports. In: none (a source). Out: output, files.

PySpark

rows = [(o["Key"], o["Size"], o["LastModified"]) for o in s3.list_objects_v2(Bucket=b, Prefix=p)["Contents"]]; df = spark.createDataFrame(rows, ["path", "size_bytes", "modified"])

pandas

df = pd.DataFrame(s3.list_objects_v2(Bucket=b, Prefix=p)["Contents"])  # paginate for more than 1,000

MongoDB input: mongodb_input

Reads documents from a MongoDB collection as rows.

Ports. In: none (a source). Out: output.

PySpark

df = spark.read.format("mongodb").option("connection.uri", uri).option("collection", coll).load()

pandas

df = pd.DataFrame(list(pymongo.MongoClient(uri)[db][coll].find(json.loads(filter), json.loads(projection))))

MySQL input: mysql_input

Reads a table, or a query's result, from MySQL or MariaDB.

Ports. In: none (a source). Out: output.

PySpark

df = spark.read.format("jdbc").option("url", "jdbc:mysql://host/db").option("query", query).load()

pandas

df = pd.read_sql(query, sqlalchemy.create_engine("mysql+pymysql://..."))

Network Share Files: smb_input

Reads files from a Windows network share (SMB).

Ports. In: none (a source). Out: output, files.

PySpark

# copy the files down with smbprotocol (smbclient), then spark.read.csv(local_paths)

pandas

with smbclient.open_file(r"\\server\share\path.csv") as f: df = pd.read_csv(f)

Parquet Files: parquet_input

Reads Parquet files uploaded to the pipeline's file library.

Ports. In: none (a source). Out: output.

PySpark

df = spark.read.parquet(*paths)

pandas

df = pd.concat([pd.read_parquet(p) for p in paths], ignore_index=True)

Postgres input: postgres_input

Reads a table, or a query's result, from PostgreSQL.

Ports. In: none (a source). Out: output.

PySpark

df = spark.read.format("jdbc").option("url", jdbc_url).option("query", query).option("user", u).option("password", p).load()

pandas

df = pd.read_sql(query, sqlalchemy.create_engine(url))

NUMERIC columns arrive as exact text in Trinium; cast them to decimal where you need numbers.

Redshift input: redshift_input

Reads a table, or a query's result, from Amazon Redshift.

Ports. In: none (a source). Out: output.

PySpark

df = spark.read.format("jdbc").option("url", "jdbc:redshift://host:5439/db").option("query", query).load()

pandas

df = pd.read_sql(query, redshift_connector.connect(**creds))

REST API (Read): rest_api_input

Reads records from a REST API, following its pages.

Ports. In: none (a source). Out: output.

PySpark

rows = []
while url:
    r = requests.get(url, headers=h, timeout=60).json()
    rows += r["data"]["items"]; url = r.get("next")
df = spark.createDataFrame(rows)

pandas

df = pd.json_normalize(rows)  # rows gathered page by page as above

S3 Files: s3_input

Reads one object, or every matching object under a folder, from Amazon S3 (or any S3-compatible store), in CSV, JSON, Parquet, Excel or XML.

Ports. In: none (a source). Out: output, files.

PySpark

df = spark.read.option("header", True).csv(f"s3a://{bucket}/{prefix}")  # .parquet(...), .json(...)

pandas

df = pd.concat([pd.read_csv(f"s3://{bucket}/{k}") for k in keys])  # needs s3fs

Bookmark has no Spark equivalent here: keep a list of processed keys yourself, or use your platform's job bookmarks. Files ending .gz are read decompressed; a .zip is read when the pattern names it.

Salesforce input: salesforce_input

Reads Salesforce records with a SOQL query, following pages.

Ports. In: none (a source). Out: output.

PySpark

recs = Salesforce(**creds).query_all(soql)["records"]; df = spark.createDataFrame(pd.DataFrame(recs).drop(columns="attributes"))

pandas

df = pd.DataFrame(Salesforce(**creds).query_all(soql)["records"]).drop(columns="attributes")  # simple-salesforce

SFTP Files: sftp_input

Reads files from an SFTP server folder.

Ports. In: none (a source). Out: output, files.

PySpark

# Spark has no SFTP reader: download with paramiko, then spark.read.csv(local_paths)

pandas

with paramiko.SSHClient() ... sftp.get(remote, local); df = pd.read_csv(local)

Trinium skips a file that is still being written (its size or time keeps changing). Reproduce that check, or read only files your partner marks as complete.

Snowflake input: snowflake_input

Reads a table, or a query's result, from Snowflake.

Ports. In: none (a source). Out: output.

PySpark

df = spark.read.format("snowflake").options(**sf_options).option("query", query).load()

pandas

df = snowflake.connector.connect(**creds).cursor().execute(query).fetch_pandas_all()

SQL Server input: sqlserver_input

Reads a table, or a query's result, from SQL Server or Azure SQL.

Ports. In: none (a source). Out: output.

PySpark

df = spark.read.format("jdbc").option("url", "jdbc:sqlserver://host;databaseName=db").option("query", query).load()

pandas

df = pd.read_sql(query, sqlalchemy.create_engine("mssql+pyodbc://..."))

SQLite Files: sqlite_input

Reads a table, or the result of a query, from an uploaded SQLite database file.

Ports. In: none (a source). Out: output.

PySpark

pdf = pd.read_sql(query, sqlite3.connect(path)); df = spark.createDataFrame(pdf)

pandas

df = pd.read_sql(query or f"SELECT * FROM {table}", sqlite3.connect(path))

Text Input: text_input

A small table typed into the pipeline itself. The columns are declared with a name and a type; the rows are the cells, by position.

Ports. In: none (a source). Out: output.

PySpark

df = spark.createDataFrame(rows, schema="region STRING, target DOUBLE")

pandas

df = pd.DataFrame(rows, columns=[c["name"] for c in columns]).astype({"target": "float64"})

Cells are text in the file; cast each column to its declared type. A cell that does not read as its type is an error in Trinium, never a silent NULL.

Trino / Starburst input: trino_input

Reads a table, or a query's result, from Trino or Starburst.

Ports. In: none (a source). Out: output.

PySpark

pdf = pd.read_sql(query, trino.dbapi.connect(host=h, port=443, user=u, http_scheme="https")); df = spark.createDataFrame(pdf)

pandas

df = pd.read_sql(query, trino.dbapi.connect(host=h, port=443, user=u, http_scheme="https"))

XML Files: xml_input

Reads XML files, turning each repeating element into a row.

Ports. In: none (a source). Out: output.

PySpark

df = spark.read.format("xml").option("rowTag", "Order").load(paths)  # Spark 4 built in, spark-xml on 3.x

pandas

df = pd.concat([pd.read_xml(p, xpath="//Order") for p in paths], ignore_index=True)

Element names are matched without namespaces.

Zendesk: zendesk_input

Reads Zendesk tickets, users or organizations.

Ports. In: none (a source). Out: output.

PySpark

# page through /api/v2/{resource}.json with requests, then spark.createDataFrame

pandas

df = pd.json_normalize(records)

Preparation

Data Cleanse: data_cleanse

Tidies text columns and drops empty rows or columns.

Ports. In: input. Out: output.

PySpark

for c in text_cols:
    df = df.withColumn(c, F.trim(F.regexp_replace(c, r"\s+", " ")))

pandas

df[text_cols] = df[text_cols].apply(lambda s: s.str.replace(r"\s+", " ", regex=True).str.strip())

Dropping all-NULL columns depends on the data, so the columns your code outputs can change from run to run.

Data Compare: data_compare

Compares two datasets on a key: one row per column on summary (types and how many values differ), and one row per differing cell, missing row or extra row on differences.

Ports. In: primary, target. Out: summary, differences.

PySpark

j = primary.alias("p").join(target.alias("t"), keys, "full_outer")
diff = j.filter(~F.col("p.amount").cast("string").eqNullSafe(F.col("t.amount").cast("string")))

pandas

m = primary.merge(target, on=keys, how="outer", suffixes=("_p", "_t"), indicator=True)
diff = m[m["amount_p"].astype(str) != m["amount_t"].astype(str)]

Values are compared as text, so a retyped column shows as one type difference rather than as every row differing. A key that is not unique is an error.

Data Profile: data_profile

Describes every column, one row per column: rows, NULLs, blanks, distinct count, min, max, mean, median, standard deviation, lengths and the top values.

Ports. In: input. Out: profile.

PySpark

df.summary().show()  # distinct: df.agg(*[F.countDistinct(c) for c in df.columns])

pandas

profile = df.describe(include="all").T

Trinium's distinct counts are exact over the whole dataset; median is approximate.

Data Validation: schema_validation

Checks rows against rules; rows that pass leave on valid, rows that fail on invalid. It can also fail the run or only warn.

Ports. In: input. Out: valid, invalid.

PySpark

ok = F.col("email").rlike(r"^[^@]+@[^@]+$") & F.col("amount").between(0, 10000)
valid_df, invalid_df = df.filter(ok), df.filter(~ok | ok.isNull())

pandas

ok = df["email"].str.match(r"^[^@]+@[^@]+$", na=False) & df["amount"].between(0, 10000)
valid_df, invalid_df = df[ok], df[~ok]

Dynamic Rename: dynamic_rename

Renames columns chosen by a rule: add or remove a prefix or suffix, replace text, change case, or compute the new name with an expression.

Ports. In: input. Out: output.

PySpark

df = df.toDF(*[re.sub(r"\W+", "_", c).lower() for c in df.columns])

pandas

df.columns = [re.sub(r"\W+", "_", c).lower() for c in df.columns]

snake case lowercases, splits camelCase and collapses punctuation to one underscore. Two columns renamed to the same name is an error.

Dynamic Select: dynamic_select

Keeps or drops columns chosen by a rule rather than by name, so new upstream columns are handled without editing.

Ports. In: input. Out: output.

PySpark

keep = [f.name for f in df.schema.fields if f.name.startswith("amt_")]
df = df.select(*keep)

pandas

df = df[[c for c in df.columns if c.startswith("amt_")]]

An empty rule matches every column. A rule that keeps nothing is an error.

Fill Missing: fill_missing

Fills NULL (and optionally blank) values per column with a fixed value or the column's mean, median or mode.

Ports. In: input. Out: output.

PySpark

mean = df.agg(F.avg("amount")).first()[0]
df = df.fillna({"amount": mean, "region": "Unknown"})

pandas

df["amount"] = df["amount"].fillna(df["amount"].mean())
df["region"] = df["region"].fillna("Unknown")

Filter: filter

Splits rows by a condition: rows where it is true leave on the true port; all others, including rows where it is NULL, leave on the false port.

Ports. In: input. Out: true, false.

PySpark

cond = F.expr("amount > 100")
true_df = df.filter(cond)
false_df = df.filter(~cond | cond.isNull())

pandas

m = df.eval("amount > 100").fillna(False).astype(bool)
true_df, false_df = df[m], df[~m]

Formula: formula

Adds or replaces columns, each computed by a SQL expression, in order: a later formula can use an earlier one's result.

Ports. In: input. Out: output.

PySpark

df = df.withColumn("total", F.expr("qty * price")).withColumn("big", F.expr("total > 1000"))

pandas

df["total"] = df["qty"] * df["price"]
df["big"] = df["total"] > 1000

Formula Multi-Field: multi_field_formula

Applies one expression to many columns, where [_current] stands for each column in turn.

Ports. In: input. Out: output.

PySpark

for c in cols:
    df = df.withColumn(c, F.expr(f"round(`{c}`, 2)"))

pandas

df[cols] = df[cols].round(2)

Formula Multi-Row: multi_row_formula

Computes a column that looks at other rows: in window mode with SQL window functions (LAG, LEAD, running totals over [_row]); in running mode with [_prev], this expression's own result on the previous row of the same group.

Ports. In: input. Out: output.

PySpark

# window mode
w = W.orderBy("_row")
df = df.withColumn("prev_amount", F.lag("amount").over(w))
# running mode needs a loop per group: use applyInPandas

pandas

# running mode, e.g. GREATEST(0, COALESCE(_prev, 0) + in - out) per group
def run(g):
    out, prev = [], None
    for a, b in zip(g["in_qty"], g["out_qty"]):
        prev = max(0, (prev or 0) + a - b); out.append(prev)
    return g.assign(balance=out)
df = df.groupby(keys, group_keys=False).apply(run)

Rows are taken in arrival order; put a Sort before it. [_row] is the row's position. [_prev] is NULL on a group's first row.

Generate Rows: generate_rows

Makes rows from nothing: a count, a number range, or a date or date-time range.

Ports. In: none (a source). Out: output.

PySpark

df = spark.range(1, count + 1).withColumnRenamed("id", "RowCount")
# dates: spark.sql("SELECT explode(sequence(DATE'2026-01-01', DATE'2026-01-31')) AS d")

pandas

df = pd.DataFrame({"RowCount": range(1, count + 1)})
# dates: pd.DataFrame({"d": pd.date_range(start_date, end_date, freq="D")})

Manage Columns: manage_columns

Keeps, drops, renames, retypes, reorders and describes columns in one place.

Ports. In: input. Out: output.

PySpark

df = df.select(F.col("cust_id").cast("bigint").alias("customer_id"), "name")

pandas

df = df.rename(columns={"cust_id": "customer_id"})[["customer_id", "name"]]
df["customer_id"] = pd.to_numeric(df["customer_id"], errors="coerce")

A value that does not convert to its new type becomes NULL (use try_cast in Spark 4, or errors="coerce" in pandas). On a text column, size truncates to that many characters. A listed column that is not in the input fails the run.

Record ID: record_id

Adds a row number, or a unique id, to every row.

Ports. In: input. Out: output.

PySpark

df = df.withColumn("RecordID", F.row_number().over(W.orderBy(F.monotonically_increasing_id())))

pandas

df.insert(0, "RecordID", range(1, len(df) + 1))

Numbers follow arrival order.

Route: route

Sends each row to the first rule it matches, one port per rule, and rows matching none to unmatched. The ports partition the input.

Ports. In: input. Out: route_1, route_2, ... (one per rule, in order), then unmatched.

PySpark

branch = F.when(F.expr(r1), 0).when(F.expr(r2), 1).otherwise(2)
df2 = df.withColumn("_route", branch)
r1_df = df2.filter("_route = 0").drop("_route")  # and so on

pandas

conds = [df.eval(r1), df.eval(r2)]
route = np.select(conds, [0, 1], default=2)
r1_df = df[route == 0]

First match wins, and a rule that is NULL for a row does not match it.

Sample: sample

Keeps part of the rows: the first N, the last N, all but the first N, every Nth, a chance per row, or N at random.

Ports. In: input. Out: output.

PySpark

df = df.limit(n)  # random: df.sample(fraction=p, seed=seed)

pandas

df = df.head(n)  # random: df.sample(n=n, random_state=seed)

Sort: sort

Sorts rows by one or more columns.

Ports. In: input. Out: output.

PySpark

df = df.orderBy(F.col("region").asc_nulls_last(), F.col("amount").desc_nulls_last())

pandas

df = df.sort_values(["region", "amount"], ascending=[True, False], na_position="last")

NULLs sort last in both directions.

Unique: unique

Splits rows into the first row of each key (unique) and the rest (duplicate).

Ports. In: input. Out: unique, duplicate.

PySpark

w = W.partitionBy(*keys).orderBy("_row")
df2 = df.withColumn("_row", F.monotonically_increasing_id()).withColumn("_n", F.row_number().over(w))
unique_df = df2.filter("_n = 1").drop("_row", "_n")
duplicate_df = df2.filter("_n > 1").drop("_row", "_n")

pandas

dup = df.duplicated(subset=keys, keep="first")
unique_df, duplicate_df = df[~dup], df[dup]

Trinium keeps the FIRST row it saw, in arrival order, and treats NULL keys as equal. Spark has no arrival order across partitions: number the rows where they are read if order matters.

Join

Append Fields: append_fields

Adds every column of the source input to every row of the target input (a cross join), typically a one-row table of constants.

Ports. In: target, source. Out: output.

PySpark

df = target.crossJoin(F.broadcast(source))

pandas

df = target.merge(source, how="cross", suffixes=("", "_source"))

Find & Replace: find_replace

Looks up each value of a column in a lookup table and replaces the matching text, or appends the lookup row's columns.

Ports. In: input, lookup. Out: output.

PySpark

m = {r["find"]: r["repl"] for r in lookup.collect()}
expr = F.col("text")
for k, v in m.items():
    expr = F.regexp_replace(expr, re.escape(k), v)
df = df.withColumn("text", expr)

pandas

m = dict(zip(lookup["find"], lookup["repl"]))
df["text"] = df["text"].replace(m, regex=False)  # entire match; use str.replace for parts

Fuzzy Match: fuzzy_match

Finds rows that are probably the same thing although written differently. Dedupe groups similar rows of one input; Match pairs each input row with similar rows of a reference input.

Ports. In: input (and reference, in match mode). Out: output in dedupe mode (with group_id, is_primary, match_score); matched and unmatched in match mode.

PySpark

@F.pandas_udf("double")
def score(a: pd.Series, b: pd.Series) -> pd.Series:
    return pd.Series([fuzz.ratio(x, y) for x, y in zip(a, b)])
# block first: join on the match key, then score pairs and keep score >= threshold

pandas

from rapidfuzz import fuzz, process
pairs = [(i, j, fuzz.ratio(a, b)) for i, a in df["name"].items() for j, b in ref["name"].items()]

Trinium's scores are its own (normalized Levenshtein, Jaro-Winkler, Soundex and Metaphone, token sets, weighted and scaled to 100). A Python library gives close but not identical scores, so expect some borderline pairs to differ. Always block on a key before comparing.

Join: join

Joins two inputs on key columns. Matched rows leave on join; unmatched left rows on left; unmatched right rows on right. Wire join alone for an inner join, join and left for a left join.

Ports. In: left, right. Out: left, join, right.

PySpark

cond = [left["id"] == right["cust_id"]]
join_df = left.join(right, cond, "inner")
left_only = left.join(right, cond, "left_anti")
right_only = right.join(left, [right["cust_id"] == left["id"]], "left_anti")

pandas

m = left.merge(right, left_on="id", right_on="cust_id", how="outer", indicator=True, suffixes=("", "_right"))
join_df = m[m["_merge"] == "both"]
left_only = m[m["_merge"] == "left_only"][left.columns]

NULL keys never match. If the file wires join and left into a Union, that is a left join: write it as one LEFT JOIN. pandas matches NaN keys to each other, so drop or fill them first.

Join Multiple: join_multiple

Joins 2 to 16 inputs on their key columns at once: a full outer join by default, or inner only.

Ports. In: input_1, input_2, ... (one per input). Out: output.

PySpark

out = dfs[0]
for d in dfs[1:]:
    out = out.join(d, on="id", how="full_outer")

pandas

out = functools.reduce(lambda a, b: a.merge(b, on="id", how="outer"), dfs)

The key comes out as one coalesced column.

Union: union

Stacks 2 to 16 inputs into one table, matching columns by name (default) or by position.

Ports. In: input_1, input_2, ... (one per input). Out: output.

PySpark

out = functools.reduce(lambda a, b: a.unionByName(b, allowMissingColumns=True), dfs)

pandas

out = pd.concat(dfs, ignore_index=True)  # join="inner" for common columns

Lookup

API Batch Lookup: rest_api_batch

Sends rows to a REST API many at a time and adds the answer for each row as new columns: one answer per row, in the order sent.

Ports. In: input. Out: output.

PySpark

def score(pdf: pd.DataFrame) -> pd.DataFrame:
    out = []
    for i in range(0, len(pdf), 100):
        rows = pdf.iloc[i:i + 100].to_dict("records")
        r = requests.post(url, json={"rows": rows}, headers=h, timeout=60)
        r.raise_for_status()
        out += r.json()["rows"]
    return pdf.assign(score=[a["score"] for a in out])
df = df.groupBy(F.spark_partition_id()).applyInPandas(score, schema=df.schema.add("score", "double"))

pandas

answers = []
for i in range(0, len(df), 100):
    rows = df.iloc[i:i + 100].to_dict("records")
    r = requests.post(url, json={"rows": rows}, headers=h, timeout=60)
    r.raise_for_status()
    answers += r.json()["rows"]
df["score"] = [a["score"] for a in answers]

This is how a pipeline calls custom code or a hosted model in bulk: the code runs behind an HTTPS endpoint you host. Check that the answer holds exactly one entry per row.

API Lookup: rest_api_call

Calls a REST API once for every row and adds the response's fields to the row.

Ports. In: input. Out: output.

PySpark

@F.pandas_udf("string")
def call(ids: pd.Series) -> pd.Series:
    return ids.map(lambda i: requests.get(f"{base}/customers/{i}", headers=h, timeout=30).text)
df = df.withColumn("resp", call("id"))

pandas

df["resp"] = df["id"].map(lambda i: requests.get(f"{base}/customers/{i}", headers=h, timeout=30).json())

This is also how a pipeline calls custom code: the code runs behind an HTTPS endpoint you host.

Database Lookup: db_lookup

Looks up each row's key in a database table or query: rows with matches leave on matched with the database's columns added; rows without, on not_found.

Ports. In: input. Out: matched, not_found.

PySpark

lk = spark.read.format("jdbc").option("url", jdbc_url).option("dbtable", table).load()
matched = df.join(lk, df["id"] == lk["cust_id"], "inner")
not_found = df.join(lk, df["id"] == lk["cust_id"], "left_anti")

pandas

keys = df["id"].dropna().unique().tolist()
lk = pd.read_sql("SELECT * FROM t WHERE cust_id = ANY(%(k)s)", engine, params={"k": keys})
matched = df.merge(lk, left_on="id", right_on="cust_id")

The database decides whether two keys match, by its own collation: 'abc' may equal 'ABC'. A NULL key goes to not_found.

Parse

DateTime: datetime

Converts text to a date or date-time using a format, or formats a date as text.

Ports. In: input. Out: output.

PySpark

df = df.withColumn("d", F.to_date("txt", "dd/MM/yyyy"))  # Spark patterns

pandas

df["d"] = pd.to_datetime(df["txt"], format="%d/%m/%Y", errors="coerce")

Translate the format: yyyy is %Y, MM %m, dd %d, HH %H, mm %M, ss %S, Mon %b, Month %B, tt %p. Spark's patterns happen to use the same letters for most tokens; Mon is MMM and Month MMMM there.

JSON: json_parse

Reads JSON held in a text column: extract chosen paths into columns, list every leaf as name and value rows, or flatten everything into columns.

Ports. In: input. Out: output.

PySpark

schema = "struct<customer:struct<id:bigint>,items:array<struct<sku:string>>>"
df = df.withColumn("j", F.from_json("payload", schema)).select("*", "j.customer.id")

pandas

parsed = df["payload"].map(json.loads)
out = pd.json_normalize(parsed)

A row whose cell holds no records still comes out, with NULLs, so the row's other columns are not lost.

RegEx: regex

Uses a regular expression on a text column: test it (match), replace it, parse capture groups into columns, or split it into tokens.

Ports. In: input. Out: output.

PySpark

df = df.withColumn("code", F.regexp_extract("text", r"([A-Z]{3})-(\d+)", 1))

pandas

df[["code", "num"]] = df["text"].str.extract(r"([A-Z]{3})-(\d+)")

Trinium's regular expressions have no look-around and no back-references. Python's re and Java's (Spark) accept more, so a pattern that works in Trinium works there; replacement groups are \1 in Python and $1 in Spark.

Text to Columns: text_to_columns

Splits a text column on delimiters into several columns, or into one row per piece.

Ports. In: input. Out: output.

PySpark

parts = F.split("tags", ",")
df = df.select("*", *[parts[i].alias(f"tag_{i+1}") for i in range(3)])  # rows: F.explode(parts)

pandas

df[["tag_1", "tag_2", "tag_3"]] = df["tags"].str.split(",", n=2, expand=True)  # rows: df.explode

URL: url_parse

Splits a URL column into parts (scheme, host, domain, path, query and so on), or turns its query string into columns or rows.

Ports. In: input. Out: output.

PySpark

df = df.withColumn("host", F.expr("parse_url(url, 'HOST')")).withColumn("q", F.expr("parse_url(url, 'QUERY', 'utm_source')"))

pandas

p = df["url"].map(urllib.parse.urlsplit)
df["host"] = p.map(lambda u: u.hostname)

Domain and suffix use the public suffix list (example.co.uk splits as example and co.uk). tldextract in Python does the same; Spark's parse_url does not.

XML: xml_parse

Reads XML held in a text column, with the same three modes as XML Files: columns, name_value or extract.

Ports. In: input. Out: output.

PySpark

df = df.withColumn("amt", F.xpath_string("xml", F.lit("Doc/Amt")))  # or from_xml in Spark 4

pandas

df["amt"] = df["xml"].map(lambda s: ET.fromstring(s).findtext("Amt"))

Transform

Cross Tab: cross_tab

Pivots: one column per distinct value of a header column, filled with an aggregate of a value column.

Ports. In: input. Out: output.

PySpark

df = df.groupBy("region").pivot("month").agg(F.sum("amount"))

pandas

df = df.pivot_table(index="region", columns="month", values="amount", aggfunc="sum").reset_index()

The new columns come from the data, so they change when the data does.

Summarize: summarize

Groups rows and aggregates them: sum, count, distinct count, min, max, average, median, standard deviation, variance, first, last, concatenate and percentile, each optionally only for rows matching a condition.

Ports. In: input. Out: output.

PySpark

df = df.groupBy("region").agg(F.sum("amount").alias("amount_sum"), F.count(F.when(F.expr("amount > 100"), 1)).alias("big"))

pandas

df = df.groupby("region", dropna=False).agg(amount_sum=("amount", "sum")).reset_index()

Output is ordered by the group columns. A NULL group key is its own group (pandas needs dropna=False). count counts rows; count_non_null counts values.

Transpose: transpose

Unpivots: turns chosen columns into rows of name and value, keeping the key columns.

Ports. In: input. Out: output.

PySpark

df = df.unpivot(ids=["id"], values=["q1", "q2"], variableColumnName="Name", valueColumnName="Value")  # Spark 3.4+

pandas

df = df.melt(id_vars=["id"], value_vars=["q1", "q2"], var_name="Name", value_name="Value")

Output

Apache Iceberg output: iceberg_output

Writes rows to an Apache Iceberg table through its catalog, as one snapshot per run.

Ports. In: input. Out: none (it writes outside the pipeline).

PySpark

df.writeTo(f"{catalog}.{namespace}.{table}").append()  # overwrite: .overwritePartitions() or .createOrReplace()

pandas

load_catalog("c", **props).load_table(f"{namespace}.{table}").append(pa.Table.from_pandas(df))  # pyiceberg

An append matches columns by name.

Azure Blob File: azure_blob_output

Writes a file to Azure Blob Storage, optionally split into key=value folders.

Ports. In: input. Out: none (it writes outside the pipeline).

PySpark

df.write.mode("overwrite").parquet(f"abfss://{container}@{account}.dfs.core.windows.net/{path}")

pandas

df.to_parquet(f"abfs://{container}/{path}", storage_options={...})

BigQuery output: bigquery_output

Writes rows to a BigQuery table with a load job.

Ports. In: input. Out: none (it writes outside the pipeline).

PySpark

df.write.format("bigquery").option("writeMethod", "direct").mode("append").save(table)

pandas

bigquery.Client().load_table_from_dataframe(df, table).result()

ClickHouse output: clickhouse_output

Writes rows to a ClickHouse table, creating a MergeTree table if needed.

Ports. In: input. Out: none (it writes outside the pipeline).

PySpark

df.write.format("jdbc").option("url", "jdbc:clickhouse://host:8123/db").option("dbtable", table).mode("append").save()

pandas

clickhouse_connect.get_client(**creds).insert_df(table, df)

CSV File: csv_output

Writes a CSV file the run keeps as a download.

Ports. In: input. Out: none (it writes outside the pipeline).

PySpark

df.coalesce(1).write.mode("overwrite").option("header", True).csv(out_dir)

pandas

df.to_csv(filename, index=False)

Databricks output: databricks_output

Writes rows to a Databricks table through a Unity Catalog volume and COPY INTO.

Ports. In: input. Out: none (it writes outside the pipeline).

PySpark

df.write.mode("append").saveAsTable(table)  # inside Databricks

pandas

# write Parquet to the volume with the Files API, then run COPY INTO through databricks-sql-connector

Excel Report: excel_output

Writes a styled Excel workbook the run keeps as a download: a sheet name, a table style, number formats, frozen header, optional chart.

Ports. In: input. Out: none (it writes outside the pipeline).

PySpark

df.toPandas().to_excel(filename, sheet_name="Report", index=False)

pandas

with pd.ExcelWriter(filename, engine="xlsxwriter") as w:
    df.to_excel(w, sheet_name="Report", index=False)

File Action: file_action

Copies, moves or deletes the files a file input or List Files hands it on a files wire, before the run reads them or after it succeeds or fails. It reads no rows.

Ports. In: files (a wire from another node's files port). Out: none (it writes outside the pipeline).

PySpark

for key in processed_keys:
    s3.copy_object(Bucket=b, Key=f"archive/{key}", CopySource={"Bucket": b, "Key": key})
    s3.delete_object(Bucket=b, Key=key)

pandas

for key in processed_keys:
    shutil.move(src / key, archive / key)

Do this after your job succeeds, never before it has finished reading.

GCS File: gcs_output

Writes a file to Google Cloud Storage, optionally split into key=value folders.

Ports. In: input. Out: none (it writes outside the pipeline).

PySpark

df.write.mode("overwrite").parquet(f"gs://{bucket}/{path}")

pandas

df.to_parquet(f"gs://{bucket}/{path}", index=False)

Google Sheets output: google_sheets_output

Writes rows to a tab of a Google Sheets spreadsheet.

Ports. In: input. Out: none (it writes outside the pipeline).

PySpark

ws.update([pdf.columns.tolist()] + pdf.astype(str).values.tolist())  # pdf = df.toPandas()

pandas

ws = gspread.service_account().open_by_key(sid).worksheet(tab)
ws.update([df.columns.tolist()] + df.astype(str).values.tolist())

JSON File: json_output

Writes a JSON file the run keeps as a download.

Ports. In: input. Out: none (it writes outside the pipeline).

PySpark

df.coalesce(1).write.mode("overwrite").json(out_dir)  # lines

pandas

df.to_json(filename, orient="records")  # lines=True for lines

MongoDB output: mongodb_output

Writes rows as documents to a MongoDB collection.

Ports. In: input. Out: none (it writes outside the pipeline).

PySpark

df.write.format("mongodb").option("connection.uri", uri).option("collection", coll).mode("append").save()

pandas

pymongo.MongoClient(uri)[db][coll].insert_many(df.to_dict("records"))

MySQL output: mysql_output

Writes rows to a MySQL table.

Ports. In: input. Out: none (it writes outside the pipeline).

PySpark

df.write.format("jdbc").option("url", "jdbc:mysql://host/db").option("dbtable", table).mode("append").save()

pandas

df.to_sql(table, engine, if_exists="append", index=False)

Network Share File: smb_output

Writes a file to a Windows network share.

Ports. In: input. Out: none (it writes outside the pipeline).

PySpark

with smbclient.open_file(remote, mode="w") as f: df.toPandas().to_csv(f, index=False)

pandas

with smbclient.open_file(remote, mode="w") as f: df.to_csv(f, index=False)

Parquet File: parquet_output

Writes a Parquet file the run keeps as a download.

Ports. In: input. Out: none (it writes outside the pipeline).

PySpark

df.write.mode("overwrite").option("compression", "zstd").parquet(out_dir)

pandas

df.to_parquet(filename, compression="zstd", index=False)

Postgres output: postgres_output

Writes rows to a PostgreSQL table, creating it if it does not exist.

Ports. In: input. Out: none (it writes outside the pipeline).

PySpark

df.write.format("jdbc").option("url", jdbc_url).option("dbtable", table).mode("append").save()
# upsert: write to a staging table, then INSERT ... ON CONFLICT in post SQL

pandas

df.to_sql(table, engine, if_exists="append", index=False, method="multi")

Run pre_sql before writing and post_sql after it succeeds, outside the load.

Redshift output: redshift_output

Writes rows to an Amazon Redshift table.

Ports. In: input. Out: none (it writes outside the pipeline).

PySpark

df.write.format("jdbc").option("url", "jdbc:redshift://host:5439/db").option("dbtable", table).mode("append").save()

pandas

awswrangler.redshift.to_sql(df, conn, table=table, schema=schema, mode="append")

REST API (Write): rest_api_output

Sends rows to a REST API, one request per row or many rows per request.

Ports. In: input. Out: none (it writes outside the pipeline).

PySpark

for batch in chunks(df.toPandas().to_dict("records"), 100):
    requests.post(url, json={"records": batch}, headers=h, timeout=60).raise_for_status()

pandas

for rec in df.to_dict("records"):
    requests.post(url, json=rec, headers=h, timeout=60).raise_for_status()

Never resend a write that may have reached the server: a retry can create the record twice.

S3 File: s3_output

Writes a file (CSV, JSON, Parquet or Excel) to Amazon S3, optionally split into key=value folders by column values.

Ports. In: input. Out: none (it writes outside the pipeline).

PySpark

df.write.mode("overwrite").partitionBy("region").parquet(f"s3a://{bucket}/{prefix}")
# replace_folder ~ spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")

pandas

df.to_parquet(f"s3://{bucket}/{key}", index=False)  # needs s3fs; partition_cols=[...] for folders

${date} is the day the run was queued, in the instance's time zone, not UTC.

Salesforce output: salesforce_output

Inserts, updates, upserts or deletes Salesforce records, 200 a request, reading Salesforce's verdict for every record.

Ports. In: input. Out: none (it writes outside the pipeline).

PySpark

sf = Salesforce(**creds)
res = sf.bulk.Account.upsert(df.toPandas().to_dict("records"), "Ext_Id__c", batch_size=200)

pandas

res = Salesforce(**creds).bulk.Account.insert(df.to_dict("records"), batch_size=200)
failed = [r for r in res if not r["success"]]

Salesforce answers HTTP 200 even when records failed: always read each record's success flag. Delete moves records to the Recycle Bin.

Send Email: email_output

Sends email: one message for the whole run, one per row, or one per group, with the group's rows attached as CSV or Excel. It outputs one row per message sent.

Ports. In: input. Out: delivery.

PySpark

for region, g in df.toPandas().groupby("region"):
    msg = EmailMessage(); msg["To"] = g["manager_email"].iloc[0]
    msg["Subject"] = f"Sales for {region}"; msg.add_attachment(g.to_csv(index=False), filename="sales.csv")
    smtp.send_message(msg)

pandas

with smtplib.SMTP(host, 587) as smtp:
    smtp.starttls(); smtp.login(user, pw)
    for region, g in df.groupby("region"): ...

{{ }} holds a SQL expression evaluated on the group's rows. Trinium refuses a group whose recipient resolves to two different addresses; check that in your code too.

SFTP File: sftp_output

Writes a file to an SFTP server.

Ports. In: input. Out: none (it writes outside the pipeline).

PySpark

df.toPandas().to_csv(local, index=False); sftp.put(local, remote)

pandas

df.to_csv(local, index=False); sftp.put(local, remote)  # paramiko

Trinium uploads to a temporary name and renames it, so a reader never sees a half-written file.

Snowflake output: snowflake_output

Writes rows to a Snowflake table.

Ports. In: input. Out: none (it writes outside the pipeline).

PySpark

df.write.format("snowflake").options(**sf_options).option("dbtable", table).mode("append").save()

pandas

snowflake.connector.pandas_tools.write_pandas(conn, df, table)

SQL Server output: sqlserver_output

Writes rows to a SQL Server table.

Ports. In: input. Out: none (it writes outside the pipeline).

PySpark

df.write.format("jdbc").option("url", "jdbc:sqlserver://host;databaseName=db").option("dbtable", table).mode("append").save()

pandas

df.to_sql(table, engine, if_exists="append", index=False, fast_executemany=True)

SQLite File: sqlite_output

Writes a table into a SQLite database file the run keeps as a download.

Ports. In: input. Out: none (it writes outside the pipeline).

PySpark

df.toPandas().to_sql(table, sqlite3.connect(filename), if_exists="replace", index=False)

pandas

df.to_sql(table, sqlite3.connect(filename), if_exists="replace", index=False)

Trino / Starburst output: trino_output

Writes rows to a Trino table.

Ports. In: input. Out: none (it writes outside the pipeline).

PySpark

# write the files to the table's location, or INSERT through the trino client in batches

pandas

df.to_sql(table, sqlalchemy.create_engine("trino://..."), if_exists="append", index=False, method="multi")

Checking the result

Run the Python code and the pipeline on the same input and compare their outputs. In Trinium, a Data Compare node compares two datasets on a key and lists every value that differs, every missing row and every extra row: read the Python output back into a pipeline and compare it with the pipeline's own.