Hi team I have a dynamic taslk which launches a launch plan Flyte #ask-the-community

Hi team, I have a dynamic taslk which launches a ...

varsha Parthasarathy

12/15/2022, 12:08 PM

Hi team, I have a dynamic taslk which launches a launch plan(workflow) When if one of the workflow fails, all of the other launches gets aborted, How to let flyte know not to do this? Its okay if one launch task fail, continue with the rest

Maarten de Jong

12/15/2022, 12:39 PM

You can set a failure policy for a workflow to allow it to continue if tasks fail. e.g.

@workflow(failure_policy=WorkflowFailurePolicy.FAIL_AFTER_EXECUTABLE_NODES_COMPLETE)

varsha Parthasarathy

12/15/2022, 4:16 PM

Thanks @Maarten de Jong, you mean for my workflow which has dynamic task ? My dynamic task launches 100 (example) workflows. If some step in the workflow fails, its okay for a workflow to fail, but i don’t want other 99 workflows to get “aborted”

varsha Parthasarathy

12/15/2022, 4:19 PM

Copy code

ground_truth_workflow = flytekit.LaunchPlan.get_or_create(
    name="ground_truth_workflow",
    workflow=GroundTruthOfflinePCPWorkFlow,
)

Dynamic tasks calls a launch plan.

Copy code

@flytekit.dynamic
def run:
     # Figures out number of iterations
     # ground_truth_workflow(..)

varsha Parthasarathy

12/15/2022, 4:19 PM

If one of the `ground_truth_workflow`(..) fails, current behavior all of the “`RUNNING`” workflows get

aborted

varsha Parthasarathy

12/15/2022, 4:58 PM

@Ketan (kumare3) - I get this error when i add a failure policy.

Copy code

TypeError: isinstance() arg 2 must be a type or tuple of types

varsha Parthasarathy

12/15/2022, 5:44 PM

cc @Dan Rammer (hamersaw) in case you have time 🙏

varsha Parthasarathy

12/15/2022, 6:58 PM

cc @Jay Ganbat - in case you know 🙏

Dan Rammer (hamersaw)

12/15/2022, 7:00 PM

Can you say more about where you see the error? What is the workflow def?

Dan Rammer (hamersaw)

12/15/2022, 7:00 PM

this should be specified on the workflow that calls the dynamic task

Jay Ganbat

12/15/2022, 7:01 PM

I think Maartens suggestion will work

varsha Parthasarathy

12/15/2022, 7:01 PM

Copy code

@flytekit.workflow
def GroundTruthSchedulerWorkflows(failure_policy=WorkflowFailurePolicy.FAIL_AFTER_EXECUTABLE_NODES_COMPLETE):

    <http://logger.info|logger.info>("scheduling ground truth workflows")
    scheduler_tasks.run()

This is the workflow which calls dynamic task

Copy code

@flytekit.dynamic(
    cache=False,
    cache_version="0.12.10",
    retries=1,
    requests=flytekit.Resources(mem="16Gi", cpu="1"),
    limits=flytekit.Resources(mem="16Gi", cpu="1"),
)
def run():
  ….

varsha Parthasarathy

12/15/2022, 7:02 PM

Copy code

File "/app/src/perception/scene_workflows/ground_truth/workflows/py_run_ground_truth_scheduler_workflows.binary.runfiles/com_lyft_avsoftware/src/perception/scene_workflows/ground_truth/workflows/ground_truth_scheduler_workflows.py", line 14, in <module>
  def GroundTruthSchedulerWorkflows(failure_policy=flytekit.WorkflowFailurePolicy.FAIL_AFTER_EXECUTABLE_NODES_COMPLETE) :
 File "/app/src/perception/scene_workflows/ground_truth/workflows/py_run_ground_truth_scheduler_workflows.binary.runfiles/pyreqs_requirements_py3_pypi__flytekit_1_0_1/flytekit/core/workflow.py", line 739, in workflow
  return wrapper(_workflow_function)
 File "/app/src/perception/scene_workflows/ground_truth/workflows/py_run_ground_truth_scheduler_workflows.binary.runfiles/pyreqs_requirements_py3_pypi__flytekit_1_0_1/flytekit/core/workflow.py", line 734, in wrapper
  workflow_instance.compile()
 File "/app/src/perception/scene_workflows/ground_truth/workflows/py_run_ground_truth_scheduler_workflows.binary.runfiles/pyreqs_requirements_py3_pypi__flytekit_1_0_1/flytekit/core/workflow.py", line 604, in compile
  self._input_parameters = transform_inputs_to_parameters(ctx, self.python_interface)
 File "/app/src/perception/scene_workflows/ground_truth/workflows/py_run_ground_truth_scheduler_workflows.binary.runfiles/pyreqs_requirements_py3_pypi__flytekit_1_0_1/flytekit/core/interface.py", line 188, in transform_inputs_to_parameters
  default_lv = TypeEngine.to_literal(ctx, _default, python_type=interface.inputs[k], expected=v.type)
 File "/app/src/perception/scene_workflows/ground_truth/workflows/py_run_ground_truth_scheduler_workflows.binary.runfiles/pyreqs_requirements_py3_pypi__flytekit_1_0_1/flytekit/core/type_engine.py", line 680, in to_literal
  transformer.assert_type(python_type, python_val)
 File "/app/src/perception/scene_workflows/ground_truth/workflows/py_run_ground_truth_scheduler_workflows.binary.runfiles/pyreqs_requirements_py3_pypi__flytekit_1_0_1/flytekit/core/type_engine.py", line 96, in assert_type
  if not hasattr(t, "__origin__") and not isinstance(v, t):

Dan Rammer (hamersaw)

12/15/2022, 7:07 PM

the failure policy needs to be defined in the workflow decorator rather than as an argument to the workflow

Copy code

@flytekit.workflow(failure_policy=...)

and then as i mentioned it should be used on the top-level workflow to define how that handles failures in the dynamic subtasks

Copy code

@flytekit.workflow(failure_policy=WorkflowFailurePolicy.FAIL_AFTER_EXECUTABLE_NODES_COMPLETE)
def foo:
    run()

ground_truth_workflow = flytekit.LaunchPlan.get_or_create(
    name="ground_truth_workflow",
    workflow=GroundTruthOfflinePCPWorkFlow,
)

@flytekit.dynamic(
    cache=False,
    cache_version="0.12.10",
    retries=1,
    requests=flytekit.Resources(mem="16Gi", cpu="1"),
    limits=flytekit.Resources(mem="16Gi", cpu="1"),
)
def run():
    for i in range(2):
        ground_truth_workflow()

varsha Parthasarathy

12/16/2022, 4:32 PM

i am so sorry for this 🤦 🤯 I don’t know why I made this error and couldn’t debug this! 🤦‍♀️

varsha Parthasarathy

12/16/2022, 4:32 PM

Thanks @Dan Rammer (hamersaw)

Dan Rammer (hamersaw)

12/16/2022, 4:38 PM

😄 no problem!

155 Views

Open in Slack

Previous Next