Usage Guide

metaflow debug task [OPTIONS] PATHSPEC

metaflow debug task FlowName/123/step_name/456 --metaflow-root-dir ~/notebooks/debug_task

--metaflow-root-dir ~/notebooks/debug_task

metaflow debug task FlowName/123/step_name/456 \
  --metaflow-root-dir ~/debug \
  --inspect

metaflow debug task FlowName/123/step_name/456 \
  --metaflow-root-dir ~/debug \
  --override-env my_custom_env

metaflow debug task FlowName/123/step_name/456 \
  --metaflow-root-dir ~/debug \
  --override-env-from-pathspec FlowName/100/other_step/789

metaflow debug --quiet task FlowName/123/step_name/456 --metaflow-root-dir ~/debug

path_components = pathspec.split("/")
if not path_components:
    raise CommandException("Provide either a flow, run, step or task to debug")

if len(path_components) < 2:
    # Flow name only - get latest successful run
    r = Flow(path_components[0]).latest_successful_run
    if r is None:
        raise CommandException(
            "Flow {} can only be specified if there is a successful run in the "
            "current namespace.".format(path_components[0])
        )
    path_components.append(r.id)

if len(path_components) < 3:
    # No step specified - use 'end' step
    path_components.append("end")

if len(path_components) < 4:
    # Enforce single task
    cur_task = None
    for t in r[path_components[2]]:
        if cur_task is not None:
            raise CommandException(
                "Step {} does not refer to a single task -- please specify the "
                "task unambiguously".format("/".join(path_components))
            )
        cur_task = t

~/notebooks/debug_task/
├── code_package.tar.gz          # Downloaded code package
├── <flow_files>.py              # Extracted flow code
├── _escape_trampolines/         # Python path overrides
│   └── metaflow/
├── debug_stub_generator.py      # Artifact stub generator
├── current_stub_generator.py    # Current object stub generator
├── <pathspec>_debug.py          # Debug script
└── debug.ipynb                  # Debug notebook

# Generated debug script
import sys
import os

# Add escape trampolines to path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '_escape_trampolines'))

# Import stub generators
from debug_stub_generator import generate_debug_stub
from current_stub_generator import generate_current_stub

# Create 'self' object with artifact access
self = generate_debug_stub('FlowName/123/step_name/456')

# Your step code can now be executed

from metaflow.plugins.env_escape import generate_trampolines
generate_trampolines(trampoline_dir)

cd ~/notebooks/debug_task
jupyter notebook debug.ipynb

# Access input artifacts
print(self.input['n_estimators'])
print(self.input['learning_rate'])

# Access data artifacts
print(self.features.shape)
print(self.labels.shape)

# Access any artifact from the task
print(self.index)

# Original step code
from sklearn import ensemble
from sklearn.model_selection import cross_val_score
import numpy as np

estimator = ensemble.GradientBoostingRegressor(
    n_estimators=self.input['n_estimators'],
    learning_rate=self.input['learning_rate'],
    max_depth=self.input['max_depth'],
    min_samples_split=2,
    loss='ls'
)

estimator.fit(self.features, self.labels)

mses = cross_val_score(
    estimator, self.features, self.labels,
    cv=5, scoring='neg_mean_squared_error'
)
rmse = np.sqrt(-mses).mean()

print(f"RMSE: {rmse}")

# Experiment with different min_samples_split
for min_samples in [2, 3, 5, 10]:
    estimator = ensemble.GradientBoostingRegressor(
        n_estimators=self.input['n_estimators'],
        learning_rate=self.input['learning_rate'],
        max_depth=self.input['max_depth'],
        min_samples_split=min_samples,
        loss='ls'
    )
    estimator.fit(self.features, self.labels)
    mses = cross_val_score(
        estimator, self.features, self.labels,
        cv=5, scoring='neg_mean_squared_error'
    )
    rmse = np.sqrt(-mses).mean()
    print(f"min_samples_split={min_samples}: RMSE={rmse:.4f}")

conda_command = [
    "metaflow",
    "environment",
    "--quiet",
    "create",
    "--name",
    mf_env_name,
    "--install-notebook",
    "--force",
]
conda_command.extend(mf_env)

result = subprocess.run(
    conda_command,
    check=True,
    capture_output=True,
)

# Parse the python path from stderr
for line in result.stderr.decode().split("\n"):
    if "/bin/python" in line:
        mf_python_path = line
        break

def _update_kernel_pythonpath(kernelspec_path, metaflow_root_dir):
    kernel_json_path = os.path.join(kernelspec_path, "kernel.json")
    with open(kernel_json_path, "r") as f:
        kernel_json = json.load(f)

    _ = kernel_json.setdefault("env", {})["PYTHONPATH"] = os.path.abspath(
        os.path.join(metaflow_root_dir, "_escape_trampolines")
    )

    for key, value in MetaflowCodeContent.get_env_vars_for_packaged_metaflow(
        metaflow_root_dir
    ).items():
        if key.endswith(":"):
            # Override existing value
            kernel_json["env"][key[:-1]] = value
        elif key not in kernel_json["env"]:
            kernel_json["env"][key] = value
        else:
            # Prepend to existing value
            kernel_json["env"][key] = f"{value}:{kernel_json['env'][key]}"