py2wf: a compiler from plain Python scripts to operator DAG workflows (.py → .pyt → workflow)

[carloea2]

Original Program:

import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.svm import SVC
from sklearn.tree import DecisionTreeClassifier
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import accuracy_score
from sklearn.preprocessing import StandardScaler
import matplotlib.pyplot as plt

# Load the dataset
file_path = 'C:/Users/carlo/Downloads/diabetes_migration/diabetes.csv'
data = pd.read_csv(file_path)

# Remove duplicate rows
data = data.drop_duplicates()

# Remove rows with null values
data = data.dropna()

# Print the minimum, maximum, and mean for all fields
print("Minimum values:\n", data.min())
print("\nMaximum values:\n", data.max())
print("\nMean values:\n", data.mean())

# Create a boxplot for the 'Pregnancies' field
plt.figure(figsize=(8, 6))
plt.boxplot(data['Pregnancies'], vert=False, patch_artist=True)
plt.title('Boxplot of Pregnancies')
plt.xlabel('Number of Pregnancies')
plt.show()

# Full overview of the data
data.describe()

# Separate features and target variable
X = data.drop('Outcome', axis=1)
y = data['Outcome']

# Split data into training and testing sets (80% train, 20% test)
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

scaler = StandardScaler()
X_train = scaler.fit_transform(X_train)
X_test = scaler.transform(X_test)

# Train Random Forest model
rf_model = RandomForestClassifier(random_state=42)
rf_model.fit(X_train, y_train)
rf_pred = rf_model.predict(X_test)
rf_accuracy = accuracy_score(y_test, rf_pred)
print(f"Random Forest Accuracy: {rf_accuracy:.2%}")

# Train SVM model
svm_model = SVC(random_state=42)
svm_model.fit(X_train, y_train)
svm_pred = svm_model.predict(X_test)
svm_accuracy = accuracy_score(y_test, svm_pred)
print(f"SVM Accuracy: {svm_accuracy:.2%}")

# Train Decision Tree model
dt_model = DecisionTreeClassifier(random_state=42)
dt_model.fit(X_train, y_train)
dt_pred = dt_model.predict(X_test)
dt_accuracy = accuracy_score(y_test, dt_pred)
print(f"Decision Tree Accuracy: {dt_accuracy:.2%}")

# Train Logistic Regression model
lr_model = LogisticRegression(random_state=42)
lr_model.fit(X_train, y_train)
lr_pred = lr_model.predict(X_test)
lr_accuracy = accuracy_score(y_test, lr_pred)
print(f"Logistic Regression Accuracy: {lr_accuracy:.2%}")

Manually Translated Workflow:

This tool result:

Summary

I have an MVP of python_to_workflow (under py2udf/src/main/python/python_to_workflow/): a compiler that takes an ordinary single-file Python script and turns it into a runnable operator DAG, using one simple, predictable rule:

Every statement becomes one operator. Then, optionally, a heuristic fuses N contiguous statements into one operator to bring the graph down to a human-scale workflow.

The output is a .pyt file, still plain, executable Python, but structured so each operator is a function with explicit input/output ports, and from there a workflow JSON of stock PythonUDFV2 operators that the engine accepts today.

The 1-line-1-operator form

pipeline.compile(source) emits the finest-grained form, which we call P⊥:

• A single _main() orchestrator keeps control flow visible, for/while/if stay as ordinary statements in _main.
• Every value-producing line becomes one operator function _<var>_N(ctx, live_in) -> live_out, including loop predicates and iterables.
• Ports are computed, not declared: a liveness analysis determines each operator's inputs (names read that were bound earlier) and outputs (names written that are still live afterwards). A variable rebound after a branch is the merge, no special merge nodes.
• Imports, class/function definitions, and constants go to a shared context (ctx), registered by reference so operators can read them without threading.

Example (input → P⊥ shape):

# input.py
data = load_rows()
total = 0
for r in data:
    total += r.amount
print(total)

# P⊥ .pyt (abridged)
def _data_1(ctx):            return load_rows()
def _total_2(ctx):           return 0
def _total_3(ctx, total, r): return total + r.amount
def _print_4(ctx, total):    print(total)

def _main():
    ctx = _Ctx(); ...
    data = _data_1(ctx)
    total = _total_2(ctx)
    for r in data:
        total = _total_3(ctx, total, r)
    _print_4(ctx, total)
_main()

P⊥ is deliberately fine-grained.

The grouping heuristic: N contiguous lines = 1 operator

One operator per statement is too many operators for a real canvas. The MVP coarsening rule is intentionally simple:

• A grouping maps each statement to a block id; consecutive statements sharing a block id fuse into a single operator.
• Fusing is a re-realization, not new codegen: the fused operator's body is the statements spliced together, and its ports are recomputed by the same liveness rule (inputs = reads entering the block, outputs = writes still live after the block).
• The default heuristic is a fixed chunk size N (with control-flow statements acting as natural boundaries), exposed as a knob. Contiguity in program order guarantees the fused graph stays acyclic.

So the whole MVP surface is: compile(source) → P⊥, and compile(source, group=N) → the same program as ⌈lines/N⌉-ish fused operators. Both outputs run, and both are verified against the original script.

What the MVP requires of the input

To make "one statement = one operator" well-defined, a small normalization front-end runs first (all source-to-source, behavior-preserving): comprehensions become loops, functions are inlined toward the main body, and global state is rewritten into explicit parameter/return threading so every dependency between two lines is a visible variable. Scripts using features outside that subset are rejected rather than miscompiled.

Non-goals (for this MVP)

• No data-parallel sharding decisions; each operator is a plain UDF.
• Single-file, synchronous Python only.

Questions for discussion

• I know there's already work on translating python to workflows with LLMs. Is there room for a compiler-based version too?
• Are we ok merging something this big? It's around 20k lines of compiler passes
• If there's interest, Is anyone interested in reviewing this?

[chenlica]

@carloea2 Can you include a screenshot of the generated workflow?