flytekit.task

flytekit.task(_task_function=None, task_config=None, cache=False, cache_version='', retries=0, interruptible=None, deprecated='', timeout=0, container_image=None, environment=None, requests=None, limits=None, secret_requests=None, execution_mode=ExecutionBehavior.DEFAULT)[source]

This is the core decorator to use for any task type in flytekit.

Tasks are the building blocks of Flyte. They represent users code. Tasks have the following properties

  • Versioned (usually tied to the git sha)

  • Strong interfaces (specified inputs and outputs)

  • Declarative

  • Independently executable

  • Unit testable

For a simple python task,

@task
def my_task(x: int, y: typing.Dict[str, str]) -> str:
    ...

For specific task types

@task(task_config=Spark(), retries=3)
def my_task(x: int, y: typing.Dict[str, str]) -> str:
    ...

Please see some cookbook task examples for additional information.

Parameters
  • _task_function (Optional[Callable]) – This argument is implicitly passed and represents the decorated function

  • task_config (Optional[Any]) – This argument provides configuration for a specific task types. Please refer to the plugins documentation for the right object to use.

  • cache (bool) – Boolean that indicates if caching should be enabled

  • cache_version (str) – Cache version to use. Changes to the task signature will automatically trigger a cache miss, but you can always manually update this field as well to force a cache miss. You should also manually bump this version if the function body/business logic has changed, but the signature hasn’t.

  • retries (int) – Number of times to retry this task during a workflow execution.

  • interruptible (Optional[bool]) – [Optional] Boolean that indicates that this task can be interrupted and/or scheduled on nodes with lower QoS guarantees. This will directly reduce the $/execution cost associated, at the cost of performance penalties due to potential interruptions. Requires additional Flyte platform level configuration. If no value is provided, the task will inherit this attribute from its workflow, as follows: No values set for interruptible at the task or workflow level - task is not interruptible Task has interruptible=True, but workflow has no value set - task is interruptible Workflow has interruptible=True, but task has no value set - task is interruptible Workflow has interruptible=False, but task has interruptible=True - task is interruptible Workflow has interruptible=True, but task has interruptible=False - task is not interruptible

  • deprecated (str) – A string that can be used to provide a warning message for deprecated task. Absence / empty str indicates that the task is active and not deprecated

  • timeout (Union[datetime.timedelta, int]) – the max amount of time for which one execution of this task should be executed for. The execution will be terminated if the runtime exceeds the given timeout (approximately).

  • container_image (Optional[str]) –

    By default the configured FLYTE_INTERNAL_IMAGE is used for every task. This directive can be used to provide an alternate image for a specific task. This is useful for the cases in which images bloat because of various dependencies and a dependency is only required for this or a set of tasks, and they vary from the default.

    # Use default image name `fqn` and alter the tag to `tag-{{default.tag}}` tag of the default image
    # with a prefix. In this case, it is assumed that the image like
    # flytecookbook:tag-gitsha is published alongwith the default of flytecookbook:gitsha
    @task(container_image='{{.images.default.fqn}}:tag-{{images.default.tag}}')
    def foo():
        ...
    
    # Refer to configurations to configure fqns for other images besides default. In this case it will
    # lookup for an image named xyz
    @task(container_image='{{.images.xyz.fqn}}:{{images.default.tag}}')
    def foo2():
        ...
    

  • environment (Optional[Dict[str, str]]) – Environment variables that should be added for this tasks execution

  • requests (Optional[flytekit.core.resources.Resources]) – Specify compute resource requests for your task. For Pod-plugin tasks, these values will apply only to the primary container.

  • limits (Optional[flytekit.core.resources.Resources]) – Compute limits. Specify compute resource limits for your task. For Pod-plugin tasks, these values will apply only to the primary container. For more information, please see flytekit.Resources.

  • secret_requests (Optional[List[flytekit.models.core.security.Secret]]) – Keys that can identify the secrets supplied at runtime. Ideally the secret keys should also be semi-descriptive. The key values will be available from runtime, if the backend is configured to provide secrets and if secrets are available in the configured secrets store. Possible options for secret stores are - Vault, Confidant, Kube secrets, AWS KMS etc Refer to Secret to understand how to specify the request for a secret. It may change based on the backend provider.

  • execution_mode (Optional[flytekit.core.python_function_task.PythonFunctionTask.ExecutionBehavior]) – This is mainly for internal use. Please ignore. It is filled in automatically.

Return type

Union[Callable, flytekit.core.python_function_task.PythonFunctionTask]