CMF fluent API + Ray runner [collecting feedback]

[sergey-serebryakov]

Introduction

This PR implements one possible version of what a CMF fluent API can look like. It tries to achieve the following goals:

• Remove some rarely used features from public API (such as typed parameters for pipelines and steps).
• Automatically create steps if none are present when users call fluent API (e.g., log_dataset).
• Initialize CMF in different usage contexts, for instance, retrieve initialization parameters from environment variables.
• Automatically identify artifact association with steps (input/output) in certain usage scenarios.

Example

Assuming a user has developed four functions - fetch, preprocess, train and test, the following is the example of CMF fluent API:

import cmflib.contrib.fluent as cmf

cmf.set_cmf_parameters(filename='mlmd', graph=False)
for step in (fetch, preprocess, train, test):
    with cmf.start_step(pipeline='my_pipeline', step=step.__name__):
        step()

API methods

Fluent API methods are categorized into three buckets:

• Set CMF parameters (set_cmf_parameters). These parameters control CMF initialization, and do not include information about pipelines, steps and executions.
• Start/end steps (start_step and end_step). These methods start a new pipeline step and ends currently active pipeline steps. The start_step method returns an instance of the Step class that can be used as a python context manager to automatically end steps.
• Logging methods (log_dataset, log_dataset_with_version, log_model, log_execution_metrics, log_metric and log_validation_output). These methods log input/output artifacts. When these methods accept artifact URL, users can provide a string or a Path object, e.g.:

  ds_path = _workspace / 'iris.pkl'
  with open(ds_path , 'rb') as stream: 
      dataset: t.Dict = pickle.load(stream) 
  cmf.log_dataset(ds_path , 'input') 

  All these methods will create a new step of one does not present.

Ray runner

This PR also contains an example of how CMF pipelines run on Ray clusters. This is possible since the fluent API can initialize the CMF using environment variables:

    pipeline_env = {
        'CMF_FLUENT_INIT_METHOD': 'env',
        'CMF_FLUENT_CMF_PARAMS': json.dumps({'filename': mlmd_store.as_posix(), 'graph': False}),
        'CMF_FLUENT_PIPELINE': 'iris',
        'CMF_FLUENT_STEP': None
    }

    for step in (fetch, preprocess, train, test):
        step_env = pipeline_env.copy()
        step_env['CMF_FLUENT_STEP'] = step.remote.__name__
        ref: ray.ObjectRef = step.options(runtime_env={'env_vars': step_env}).remote()
        ray.get(ref)