Reactive execution graphs: expressing event-based dependencies

[cosmicBboy]

This is an RFC draft for supporting event-based dependencies between workflows in Flyte.

User Story

As a Flyte workflow author, I want to kick-off a task/workflow based on an event so that I can establish flexible dependencies between tasks/workflows without having to compose them into yet another workflow.

Use Case: Model Re-training

Suppose I have three workflows:

• model_training: takes data/hyperparams as input, outputs a trained model
• assess_model_drift: takes the trained model and assesses it for data/concept drift using the latest data
• deploy_model: deploys model to some external serving layer

When assess_model_drift runs, it emits an output value that indicates whether the model needs to be re-trained. If this workflow returns True, the model_training workflow should be executed.

Example Code

from flytekit import workflow


@workflow
def model_training() -> Model:
    ...


@workflow
def assess_model_drift() -> bool:
    ...

@task
def deploy_model(model: Model):
    ...

@trigger(
    on_event=WorkflowExecutionEvent(
        workflow=assess_model_drift,
        phases="SUCCEEDED",
        matches_output=True,
        alias="retrain_event",
    )
)
def retraining() -> Model:
    model = model_training()
    # use gate nodes here in case we want a human-in-the-loop to
    # judge whether to re-deploy or not.
    deploy_model(model=model)

Event Types

• When an upstream workflow execution successfully terminates
• When an upstream workflow outputs a specific value (e.g. True or False)
  • This can be limited to some set of types, such as bool or str
• When upstream data lands on some blob store path (e.g. s3)

Resources

• Flyte Reactive design doc: note that this is the original RFC written for this feature.

[fg91]

without having to compose them into yet another workflow.

Devils advocate: One could achieve the same by doing exactly this in combination with a conditional, right? Why is this solution not good enough?

When upstream data lands on some blob store path (e.g. s3)

I think this is a very interesting feature. A while back I prototyped a workflow trigger that starts an execution when newly labeled data is added to a bucket. For that I used GCP pub/sub notifications that fire when such data is added and a cloud function listening to pub/sub that used flyte remote to trigger the workflow. That worked but it would be very nice if such external triggers could be implemented purely in Flyte.

[MariaSL]

Hey!
We’re having a similar story - majority of our pipelines have external dependencies and we have a need to reference those external dependencies. 
Most of our use cases fall under this category:

When upstream data lands on some blob store path (e.g. s3)

Use-case:

We have a workflow with a task (head node) depending on upstream files written to s3. These files can arrive anytime between midnight and 5am or later, latencies are not uncommon.

@task
def read_user_data(date: Optional[str] = None) -> DataFrame:
    ...

@task
def preprocess_data(user_data: DataFrame) -> DataFrame:
    ...

@task
def train(df: DataFrame) -> str:
    ...

@task
def predict(df: DataFrame, model: str) -> None:
    ...

@workflow
def wf(date: Optional[str] = None) -> int:

    user_data = read_user_data(date=date)
    preprocessed_data = preprocess_data(user_data=user_data)
    model = train(df=preprocessed_data)
    predict(df=preprocessed_data, model=model)
    
    return 1

How did we use to handle delayed external dependencies:

• we have a frequent cron schedule - multiple times per hour there will be a Luigi workflow created;
• the Luigi workflows uses ExternalTask to refer to files created externally;

If the external dependencies are not satisfied, then, depending on our retry policy (retry_external_tasks is True or False) Luigi will:

• either re-test external tasks for completion while the worker is running (as many times as possible with a 15 min delay (retry_delay)) if retry_external_tasks is True;
• or only evaluate external tasks when Luigi is first invoked (retry_external_tasks is false);
• if all external dependencies are PENDING (no jobs to run) - the worker will die out with: Did not run any tasks This progress looks :| because there were missing external dependencies.

Something to note here is that the workflow won't fail if the external dependencies are not fulfilled. Most of the time the external dependencies are just delayed and not failed. Plus, we wouldn't want to have 20+ Flyte workflows failing many-many times.

Problem:
Right now we're not able to replicate the same behaviour as above with native Flyte functionality. One workaround we thought of:

@task
 def conditional_task(date: str) -> bool:
    """Check if upstream file exists or not"""
    ...
    
@task
 def success() -> bool:
     """
     Mimic a task that handles success
     """
     return True


 @workflow
 def main_wf(date: str) -> bool:
     sensor_output = conditional_task(date=date)
     res = (
         conditional("test")
         .if_(sensor_output.is_true())
         .then(wf(date=date))
         .else_()
         .then(success())
     )
     return res
 

This solution doesn't look suitable for complex pipelines with many delayed external dependencies.

Desired behaviour:

On a schedule kick-in:

• check if the dependencies (S3 file or S3 prefix) are ready or not;
• if the dependencies are ready and the workflow is not running yet - run the workflow;
• if the dependencies are ready and there is a workflow running already - do nothing; (related to [Feature] Prevent concurrent execution  #267)
• if the dependencies are not ready - retry N times with an exponential delay and otherwise log message and do graceful exit (terminate the workflow);

The approach we're suggesting is more of a poll/pull compared to the push/event-based suggested by @cosmicBboy.
However, our main wish is to have a solution, rather than the specific one I've described :)

[davidmirror-ops]

2023-11-09 Contributor's meetup notes: more investigation is needed to confirm if using Airflow Operators enables what’s described in this entry.