Partial execution of pipelines

[akchinSTC]

Related Issues:

#1174 #2229

Ability to start/restart a pipeline from a selected node in a pipeline

Airflow

• https://airflow.apache.org/docs/apache-airflow/1.10.15/dag-run.html#re-run-tasks

Kubeflow

• Experiment policy : resume from volume - https://www.kubeflow.org/docs/components/katib/resume-experiment/#resume-policy-fromvolume
• Seems to be an expectation that the user resubmits the same pipeline after resolving any issues and the cache will decide where it start from e.g. from the last successful step

Local

• Promotion of Local execution runtime into its own full fledged runtime, on the same level as KFP or airflow
  • Related issues: platform: pipeline-local Related to usage of the local environment as pipeline runtime

[ptitzler]

Generally speaking there appear to be two potential approaches to supporting partial pipeline execution:

• Create a new pipeline on the fly that only includes the node that the user selected and its downstream nodes. This new pipeline is submitted to the target runtime environment.
• Submit the entire pipeline and exploit the target runtime's capability to use previously cached results. The assumption is here that the user has executed the pipeline before and the target runtime uses the cached results and thus skips execution of certain nodes. This approach has a major flaw. There are many scenarios where the target runtime would execute more nodes than requested, because previously cached results cannot be re-used. A user can accidentally trigger these cache misses, for example by modifying almost any generic node property, or even by changing the runtime configuration.

Initial plan (3.9):

• Support in the VPE
• Not supported by elyra-pipeline submit or elyra-pipeline run
• Add support for local runtime (JupyterLab)
• Add support for Kubeflow Pipelines
  • Investigate whether KFP step caching can be exploited to "simulate" run from
    • General idea: Elyra uploads the entire pipeline but KFP only executes those nodes that were not yet successfully executed
    • "Kubeflow Pipelines caching provides step-level output caching. And caching is enabled by default for all pipelines submitted through the KFP backend and UI. The exception is pipelines authored using TFX SDK which has its own caching mechanism. The cache key calculation is based on the component (base image (spec.name.image), command-line (spec.name.container.command?), code (?), arguments (spec.name.container.args ?) passed to the component (values or artifacts) and any additional customizations. If the component is exactly the same and the arguments are exactly the same as in some previous execution, then the task can be skipped and the outputs of the old step can be used." [source]
    • Requires caching to be not disabled - it's enabled by default
    • Might want/need to expose enable/disable cache as pipeline default/node property
• No support for Apache Airflow

[ptitzler]

Most basic test scenario:
- Create a 1 node pipeline that runs a notebook
- The node has no properties configured, aside from the required runtime image.
- Run the pipeline from the VPE. Wait for the run to complete.
- In the Kubeflow Central Dashboard clone the run. (Nothing pipeline specific changes as part of the cloning - e.g. the notebook or the configured properties.) Wait for the run to complete and note whether the cached results are used.

Preliminary test/investigation update 1:

• Caching is performed for custom components
• Caching is not performed for generic components.

There are several potential reasons that prevent caching:

• Elyra passes the ELYRA_RUN_NAME env variable, which is unique for each pipeline run.
• Elyra passes several container arguments as a string, which might differ across runs. Example of user actions that cause the string to be different:
  • User renames pipeline
  • User modifies runtime config or uses different runtime config
  • User re-submits pipeline, which alters the unique value of cos-directory
  • User changes cos_path_prefix pipeline property
• Elyra sets several several environment variables in the container
  - User adds, removes or modifies an environment variable

[akchinSTC]

Airflow checks

• Caching is performed in airflow as long as the previous DAG run pods are left on the cluster e.g. Set the is_delete_operator_pod parameter to False in your DAG k8s pod op code (set to this by default)
• This is true for both generic and custom components
• Partial pipeline execution looks like it can be accomplished

  • via the UI via DAG or task instance view. Youll need to clear the previous task run states and choose to ignore parent task dependencies along with the previous task pods (otherwise the cache will be used)
    
    

  • or CLI (investigating). See Airflow CLI reference