GitHub user carloea2 edited a discussion: py2wf: a compiler from plain Python
scripts to operator DAG workflows (.py → .pyt → workflow)
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:
<img width="1105" height="1285" alt="image"
src="https://github.com/user-attachments/assets/92c5e1e3-bfc6-4c08-bdf1-5d4e9e1e12d3";
/>
This tool result:
<img width="1254" height="1053" alt="image"
src="https://github.com/user-attachments/assets/74295e85-4c77-4d4b-98a3-1d1fe0425ef1";
/>
## 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):
```python
# input.py
data = load_rows()
total = 0
for r in data:
total += r.amount
print(total)
```
```python
# 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?
GitHub link: https://github.com/apache/texera/discussions/6618
----
This is an automatically sent email for [email protected].
To unsubscribe, please send an email to: [email protected]
