Repository navigation
Conversation
Change map
Requirement: show how Databricks, SageMaker and Snowflake schedule kumo-engine jobs with MLOps on, track the fit and hand the model to scoring. Existing: |
|
Warning Review limit reachedEnable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. Next included review available in 24 minutes. View limit details
Walkthrough
Priority: Merge Risk:
|
65608c7 to
7014f45
Compare
91d11cd to
29f8c7f
Compare
7014f45 to
7a38c4b
Compare
29f8c7f to
2aa4786
Compare
7a38c4b to
6b20cb6
Compare
2aa4786 to
937102d
Compare
6b20cb6 to
10c4a23
Compare
937102d to
dd4e481
Compare
10c4a23 to
573186e
Compare
dd4e481 to
7b51c15
Compare
573186e to
62b64b6
Compare
7b51c15 to
cb68ce2
Compare
62b64b6 to
04f3ea1
Compare
cb68ce2 to
e203171
Compare
04f3ea1 to
913d621
Compare
e203171 to
af92e85
Compare
6bfa01e to
19bd626
Compare
56b8448 to
bb2b95a
Compare
19bd626 to
d8a7093
Compare
bb2b95a to
57045b8
Compare
d8a7093 to
caa5017
Compare
57045b8 to
da8b13f
Compare
da8b13f to
3fe4c18
Compare
e07a332 to
b3dea5a
Compare
2bfe866 to
f380a89
Compare
b3dea5a to
2c18051
Compare
f380a89 to
f21cf92
Compare
f21cf92 to
f258bb5
Compare
There was a problem hiding this comment.
Actionable comments posted: 1

Review info
-
Run configuration-
- Source mode: Auto
- Repository: NVIDIA/kumo-engine
- Branch: vedaant/mlops-mlflow-model
- File: .coderabbit.yaml
- Commit: 9b8a83a
- Layers: Repository YAML
- Review profile: QUIET
- Plan: Enterprise
- Run ID:
fa4abf28-62b5-4f3c-95ba-3c9df2d4ea74
Files selected for processing: 13-
examples/mlops/README.mdexamples/mlops/databricks/README.mdexamples/mlops/databricks/databricks.ymlexamples/mlops/fit_job.pyexamples/mlops/jobs.pyexamples/mlops/sagemaker/README.mdexamples/mlops/sagemaker/pipeline.pyexamples/mlops/score_job.pyexamples/mlops/snowflake/README.mdexamples/mlops/snowflake/register_jobs.pypackages/kumo-engine/kumo/engine/connector/snowflake.pypackages/kumo-engine/test/test_job_templates.pypackages/kumo-engine/test/test_snowflake_connector.py
Included review availability: This review used your included allowance. Your plan provides up to 12 included reviews per hour; 0 remain after this review.
-
| arguments=[ | ||
| *flags, '--scratch-dir', '/opt/ml/processing/scratch' | ||
| ], |
There was a problem hiding this comment.
Persist the processing step’s predictions.
Predict passes no --predictions-table or other output destination. score_job.main() therefore calls trainer.predict() without an output configuration, prints only a row count, and exits. The pipeline can report success without retaining any predictions. Add a supported persistent output destination to this step. (docs.aws.amazon.com)
Teams want to run kumo-engine fits and scoring on the scheduler they already have, but there is no runnable example of wiring the engine, MLOps settings and a registry into a scheduled job. examples/mlops holds two plain Python job scripts (fit_job.py, score_job.py) and a shared jobs.py. They take CLI flags (--artifact-root, --source, --registry, --model-name, --model-path, --predictions-table, ...), build kumo.MLOps from them and run inside `with kumo.mlops(cfg)`, so snapshots, training tables and models are stored under the artifact root (models/<bundle id>/). With a registry the fit registers the model and moves `champion`; scoring loads it, or --model-path from the root. Platform wiring: a Databricks Asset Bundle, a SageMaker Pipeline and (experimental) Snowflake ML Jobs + Tasks, each passing the flags. SnowflakeConnector.session() exposes the connector's Snowpark session for SnowflakeTracking. Verified with test_job_templates.py (local fit then score) and test_snowflake_connector.py. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Series: kumo-engine MLOps series: every snapshot, training table, model and prediction run carries a self-contained _provenance.json;
with kumo.mlops(...)stores each artifact under one artifact root, tracks fits and prediction runs in MLflow or Snowflake and registers models; volumes and stages can hold files; prediction runs are monitored for drift and realized performance.Stack: Models, registry and job templates: #57 → #59 → #65 → #71
Merge order: one linear stack; merge bottom to top (#53 first)
This PR: Adds job templates that run kumo-engine on Databricks Jobs, SageMaker Pipelines, and Snowflake ML Jobs with Tasks.
Summary
This PR adds job templates that run kumo-engine on Databricks Jobs, SageMaker Pipelines, and Snowflake ML Jobs with Tasks. Two plain Python job scripts take CLI flags and run inside
with kumo.mlops(kumo.MLOps(...)), so every artifact lands under one artifact root. Per-platform wiring schedules them, passes the flags, tracks the fit, and hands the trained model to the scoring job: by registry alias, or by itsmodels/<bundle id>path without a registry.Problem. A team that wants to retrain weekly and score daily can't answer these questions today:
Today none of these can be answered without reading engine source.
Cause. The repository ships only a local SDK example:
Change.
kumo_train:fit_job.pyon a single-node CPU job clusterkumo_score: dailyscore_job.pyinto a UC table (created paused)--registry mlflow:MLOps(tracking=MlflowTracking(), registry=ModelRegistryLocation("<catalog>.<schema>.users_churn", alias="champion"))(#65); scoring loadsmodels:/…@championFittraining stepPredictprocessing step--registry mlflowagainst SageMaker managed MLflow; scoring loadsmodels:/users-churn@championKUMO_TRAIN: weeklyKUMO_SCORE: daily into a table--registry snowflake:MLOps(tracking=SnowflakeTracking(...), registry=ModelOutputConfig(sf, location=ModelRegistryLocation("USERS_CHURN", alias="champion")))(#59); scoring loadschampionpython fit_job.pypython score_job.pymodel: models/<bundle id>; scoring loads--model-path models/<bundle id>from the artifact rootkumo.MLOpsfrom its flags and runs insidewith kumo.mlops(cfg). The snapshot, the training table and the model are stored under--artifact-rootwhen they are produced (Export and load graph snapshots through any file destination #54, Record training-table lineage across export and reload #55, Record bundle_id and fit provenance in every model bundle #56). The fit is tracked and the model registered when--registryis given. Scoring writes--predictions-tablewhen given.--artifact-root,--source(databricks,snowflakeor a Parquet directory),--catalog/--schema/--warehouse,--registry,--model-name,--model-path,--snapshot,--predictions-table,--query,--num-experimentsand--scratch-dir. Credentials come from the platform identity. Only MLflow's own settings (MLFLOW_TRACKING_URI, ...) stay environment variables.trainer.result.scoring_defaults). The template query (PREDICT COUNT(orders.*, 0, 30, days) > 0 FOR EACH users.USER_ID) is binary, and a binary predict without a threshold fails.spark_python_taskwhoseparametersare the flags. The artifact root is/Volumes/<catalog>/<schema>/<volume>/kumo, and the cluster env setsMLFLOW_TRACKING_URI=databricksandMLFLOW_REGISTRY_URI=databricks-uc.requirementsfile, then the kumo-engine wheel. The file starts with--prefer-binary, because DBR 16.4's setuptools cannot build the SQLAlchemy sdist MLflow pulls.spark_python_taskruns the script without__file__, so the scripts find their directory fromsys.argv[0].container_argumentsand the processing step asarguments.MLFLOW_TRACKING_URIis the tracking-server or MLflow App ARN, throughsagemaker-mlflow.--instance-typeis required.MLJobDefinition.registerdefines the train and score jobs. Each Task runsMLJobDefinition.to_sql(job_args=<flags>, use_async=False)directly; the oldtask.sqlexisted only to pass job ids.SnowflakeConnector.session().s3://orgs://URI that the external access integration reaches.Real output of
fit_job.pythenscore_job.pyon a workstation, CPU, no registry. Data:--source <data dir>with synthetic users/items/orders (50 users, 20 items, 3,000 orders over 120 days), a 7-day query, one experiment. Hashes shortened:Back-compat
Examples plus one additive connector method.
SnowflakeConnector.session()test_job_templates.pyskips whenexamples/mlopsis absent.Known limitations
champion, and scoring loaded it. This flag-based revision is verified locally only. The Snowflake template has not been run and is marked experimental.--predictions-table,score_job.pystores nothing yet; the prediction-run store underpredictions/<id>/arrives with Record a manifest and output score statistics for every prediction run #67.SYSTEM$EXECUTE_ML_JOB, whichMLJobDefinition.to_sqlgenerates, is not a documented public API.Testing Done
test_job_templates.py:fit_job.pythenscore_job.pyend to end on local Parquet (no registry): the fit printsmodel: models/<id>, the root holdsmodels/<id>/_provenance.json,snapshots/…andtraining_tables/…, and the score loads--model-path models/<id>and scores every user;--artifact-rootand--registry mlflowto both scripts, and--predictions-tableto the score job only;to_sqlstatement with the flags (--predictions-tableon the score job only);__file__(Databricksspark_python_task);--instance-type, and SageMaker and Snowflake name the scripts;examples/mlopsbyte-compiles.test_snowflake_connector.py:session()is the reader's session for the connector's target.test_job_templates.py+test_snowflake_connector.pygave 20 passed. flake8 is clean onexamples/mlops.Checklist
🤖 Generated with Claude Code