Define Flows

Flow definitions are defined using Flyte’s Python library.

Flow files

Workflow basics

Flows are defined inside a Python file. In this document, we call the file workflow.py, but the name can be anything.

workflow.py

from flytekit import workflow

@workflow
def training_workflow():
    return

We’re using the @workflow decorator to declare a function as a flow. The method name, training_workflow, is implicitly used as the name of the flow.

Inputs & outputs

Your flow can be parameterized so that you can re-run it with new inputs without modifying the code of your flow. Inputs can be either structured data or files.

Files

Declare file inputs and outputs using the generic FlyteFile type.To enable file rendering in the UI, this type must either:

Be combined with a TypeVar that defines the file extension type, for example, FlyteFile[TypeVar("csv")] and FlyteFile[TypeVar("pdf")].
Be a Flow Artifact like Artifact(name="My Data", type=DATA).File(name="data.csv")

Python base types

Use base Python types for passing structured data through flows:str, bool, int, float, list, dict, datetime

Data science types

A number of common data science types are supported as well:np.ndarray, pandas.DataFrame, pyspark.DataFrame, torch.Tensor / torch.nn.Module, sklearn.base.BaseEstimator, tf.keras.Model

Explicit, strongly typed inputs and outputs ensure that tasks pass data to each other correctly. For instance, a data processing task may emit a Pandas dataframe as output, while a training task accepts a Pandas dataframe as input. Flows analyze workflow submissions to prevent user error before execution, providing actionable feedback for incorrectly specified parameters that don’t satisfy task contracts.

Input and output types are also used for caching, data lineage tracking, and previewing in Domino.

Here we parameterize our flow by accepting a parameter (n) to our @workflow function and returning one of our results (the "results" output in sim):

workflow.py

from flytekit import workflow
from flytekit.types.file import FlyteFile
from typing import TypeVar, NamedTuple

@workflow
def simulation(n: int) -> FlyteFile[TypeVar("csv")]:

    # Simulate data using workflow parameter
    sim = run_domino_job_task(
        flyte_task_name="Simulate data",
        command="python sim-data.py",
        inputs=[Input(name="n", type=int, value=n)],
        output_spec=[Output(name="results", type=FlyteFile[TypeVar("csv")])],
        use_project_defaults_for_omitted=True
    )

    return sim["results"]

Flow tasks

Declare tasks

Tasks are the core building blocks within a flow and are isolated within their own container during an execution. A task maps to a single Domino Job.

Flow Task Jobs vs standalone Jobs

While all tasks trigger a unique Domino Job, there are some differences between jobs launched by a flow and standalone jobs. More specifically, for jobs launched by a flow:

Only Domino Dataset snapshots and NetApp Volume snapshots can be mounted, unlike standalone Domino Jobs which support snapshot mounts and read-write mounts. For data to be used in a flow, it must be part of a versioned Dataset snapshot (version 0 of a dataset, that is, the read-write directory is NOT considered a snapshot), or part of a NetApp Volume snapshot.
Only one snapshot of a Dataset or NetApp Volume may be mounted at a time. When leveraging the use_latest flag, the latest snapshot is mounted.
Snapshots are read-only and cannot be modified during a job. Any processed data that needs to be persisted should be defined and written as a task output and therefore written to the Flow blob storage.
Snapshots are mounted to a standard location that doesn’t include a snapshot ID in the path (the same location used for the latest version of a dataset or NetApp Volume in a workspace). For DFS projects, the path is:
- /domino/datasets/local/{name} for local dataset snapshots.
- /domino/datasets/{name} for shared dataset snapshots.
- /domino/netapp-volumes/{name} for NetApp Volume snapshots.
For Git-based projects, the path is:
- /mnt/data/{name} for local dataset snapshots.
- /mnt/imported/data/{name} for shared dataset snapshots.
- /mnt/netapp-volumes/{name} for NetApp Volume snapshots.
Snapshots of the project code and artifacts (results) are not taken at the end of the job. Any results that need to be persisted should be defined and written as a task output and therefore written to the Flow blob storage.
There are two additional directories: /workflow/inputs and /workflow/outputs. These are where the task inputs/outputs of the flow are stored. A job status is considered failed if the expected task outputs are not produced by the end of execution. See the Write task code section for more details on how to write your code accordingly.
Stopping a job orchestrated by a Flow in the Jobs UI stops the entire flow, including other jobs that are running as part of it.
Completed jobs cannot be re-run through the Domino Jobs UI. They must be relaunched by re-running the task from the Flows UI.
Additional job metadata is captured and displayed in the Job Details to reference the flow and task that launched it.

These differences help to guarantee the reproducibility of flow executions by ensuring the triggered jobs adhere to a strict contract and remain side-effect free.

Tasks for a flow can be defined in one of the following ways:

Use the base DominoJobConfig and DominoJobTask classes. These provide more flexibility and direct control of the exact parameters to use.
Use the run_domino_job_task helper method. This offers a more user-friendly abstraction that enables definition and execution of a task in the same line of code.

@workflow
def training_workflow(data_path: str) -> FlyteFile[TypeVar("csv")]:

    # Define our data prep task
    data_prep_job_config = DominoJobConfig(Command="python prep-data.py")
    data_prep_job = DominoJobTask(
        name='Prep data',
        domino_job_config=data_prep_job_config,
        inputs={'data_path': str},
        outputs={'processed_data': FlyteFile[TypeVar("csv")]},
        use_latest=True
    )

    # Get our task results
    data_prep_results = data_prep_job(data_path=data_path)

    return data_prep_results["processed_data"]

@workflow
def training_workflow(data_path: str) -> FlyteFile[TypeVar("csv")]:

    # Define our data prep task
    data_prep_results = run_domino_job_task(
        flyte_task_name="Prepare data",
        command="python prep-data.py",
        inputs=[Input(name="data_path", type=str, value=data_path)],
        output_spec=[Output(name="processed_data", type=FlyteFile[TypeVar("csv")])],
        use_project_defaults_for_omitted=True
    )

    return data_prep_results["processed_data"]

In both cases, tasks trigger a Domino Job with the specified settings and return the results.

Pass data

It’s common for one task to depend on another task; that is, one task accepts an input that is produced by another task as an output. This ensures that the dependent task does not start execution until outputs from the other task are produced first. To create dependent tasks, you can use either the base classes or helper methods to define them. In the example below, note how the second task uses the output from the first task by calling data_prep_results["processed_data"].

@workflow
def training_workflow(data_path: str):

    # Define our data prep task
    data_prep_job_config = DominoJobConfig(Command="python prep-data.py")
    data_prep_job = DominoJobTask(
        name='Prep data',
        domino_job_config=data_prep_job_config,
        inputs={'data_path': str},
        outputs={'processed_data': FlyteFile[TypeVar("csv")]},
        use_latest=True
    )

    # Define our training task
    training_job_config = DominoJobConfig(Command="python train-model.py")
    training_job = DominoJobTask(
        name='Train model',
        domino_job_config=training_job_config,
        inputs={'processed_data': FlyteFile[TypeVar("csv")]},
        outputs={'model': FlyteFile[TypeVar("pt")]},
        use_latest=True
    )

    # Get our task results
    data_prep_results = data_prep_job(data_path=data_path)
    training_results = training_job(processed_data=data_prep_results["processed_data"])

    return

@workflow
def training_workflow(data_path: str):

    # Define our data prep task
    data_prep_results = run_domino_job_task(
        flyte_task_name="Prepare data",
        command="python prep-data.py",
        inputs=[Input(name="data_path", type=str, value=data_path)],
        output_spec=[Output(name="processed_data", type=FlyteFile[TypeVar("csv")])],
        use_project_defaults_for_omitted=True
    )

    # Define our training task
    training_results = run_domino_job_task(
        name="Train model",
        command="python train-model.py",
        inputs=[Input(name="data", type=FlyteFile[TypeVar("csv")], value=data_prep_results["processed_data"])],
        outputs=[Output(name="model", type=FlyteFile[TypeVar("pt")])]
    )

    return

Return outputs

You can set the output of the flow by returning it in the method. Note that defining an overall Flow output is not required and does not elevate this particular output in the UI. Please see Define Flow Artifacts for elevating important task outputs.

@workflow
def training_workflow(data_path: str) -> FlyteFile[TypeVar("pt")]:

    # Define our data prep task
    data_prep_job_config = DominoJobConfig(Command="python prep-data.py")
    data_prep_job = DominoJobTask(
        name='Prep data',
        domino_job_config=data_prep_job_config,
        inputs={'data_path': str},
        outputs={'processed_data': FlyteFile[TypeVar("csv")]},
        use_latest=True
    )

    # Define our training task
    training_job_config = DominoJobConfig(Command="python train-model.py")
    training_job = DominoJobTask(
        name='Train model',
        domino_job_config=training_job_config,
        inputs={'processed_data': FlyteFile[TypeVar("csv")]},
        outputs={'model': FlyteFile[TypeVar("pt")]},
        use_latest=True
    )

    # Get our task results
    data_prep_results = data_prep_job(data_path=data_path)
    training_results = training_job(processed_data=data_prep_results["processed_data"])

    return training_results["model"]

@workflow
def training_workflow(data_path: str) -> FlyteFile[TypeVar("pt")]:

    # Define our data prep task
    data_prep_results = run_domino_job_task(
        flyte_task_name="Prepare data",
        command="python prep-data.py",
        inputs=[Input(name="data_path", type=str, value=data_path)],
        output_spec=[Output(name="processed_data", type=FlyteFile[TypeVar("csv")])],
        use_project_defaults_for_omitted=True
    )

    # Define our training task
    training_results = run_domino_job_task(
        name="Train model",
        command="python train-model.py",
        inputs=[Input(name="data", type=FlyteFile[TypeVar("csv")], value=data_prep_results["processed_data"])],
        outputs=[Output(name="model", type=FlyteFile[TypeVar("pt")])]
    )

    return training_results["model"]

Write task code

Domino code execution aims for consistency across all execution modalities. To support this, flows can leverage the same data mount paths used in a workspace or standalone job. As an input, declared inputs that reside under /mnt are populated for reading. Data written back to data paths can optionally be used to create a new snapshot of the data at the end of the flow run. However, for explicit inputs and outputs, flows additionally mounts data for input and output in two special directories under /workflow. Files can be read from /workflow/inputs and written to /workflow/outputs. In addition to files, flows also allows for the passing of structured data such as dataframes and lists. These objects also show up in the /workflow directories, but require special handling to read them as structured data in Python.

Read inputs

For each input that is defined for a task, a file is created and accessible within a Job within /workflow/inputs/ in a file whose name exactly matches that of the named input. For file type inputs, the file can be read directly at the given path.

named_input = "processed_data"
data_path = "/workflow/inputs/{}".format(named_input)
df = pd.read_csv(data_path)

named_input <- "processed_data"
df <- read.csv(file.path("/workflow", "inputs", named_input))

For Python non-file types (str, bool, int, list, dict, etc.), the blob contents contain the input value. Example usage:

input_name = "data_path"
input_location = f"/workflow/inputs/{input_name}"
with open(input_location, "r") as file:
    input_value = file.read()

Show Configuring inputs for local task execution

The Domino Python helper library has functions that facilitate setting up input data. These functions allow you to use inputs or outputs from past Flyte executions.
Here is an example that shows setting up local input data for a task that expects five inputs:

my_csv_one, using the output of a previous Flow execution.
my_csv_two, using the input of a previous Flow execution.
my_csv_three, using the local file test_data.csv.
my_str, using the string "my string input data".
my_int, using the integer 42.

import shutil
from flytekitplugins.domino.helpers import BlobDataLocation, PrimitiveDataLocation, setup_workflow_data

blobs = [
    BlobDataLocation(
        # you can find blob URLs by inspecting node executions in the Domino Flyte UI
        "s3://flyte-data-bucket/past/execution1/output/some_csv",
        # the input name will incorrectly be inferred as "some_csv" from the blob url,
        # so the local_filename kwarg must be provided and match the input name expected by the task code
        local_filename="my_csv_one",
    ),
    BlobDataLocation("s3://flyte-data-bucket/past/execution2/input/my_csv_two"),
]

primitives = [
    PrimitiveDataLocation("my string input data", "my_str"),
    PrimitiveDataLocation(42, "my_int")
]

setup_workflow_data(blobs, primitives)
shutil.copyfile("test_data.csv", "/workflow/inputs/my_csv_three")

Write outputs

Outputs defined for a task must be written into /workflow/outputs/ to a file that exactly matches the name of the output.

named_output = "processed_data"
df.to_csv("/workflow/outputs/{}".format(named_output))

named_output <- "processed_data"
write.csv(df, file.path("/workflow", "outputs", named_output))

If the defined outputs do not exist at the end of a Domino Job the task fails.

Flow jobs do not make commits or dataset snapshots. Results should always be written as outputs. Snapshots are taken at the end of flow execution if configured.

Domino Flyte plugin

Domino provides a set of Flyte-compatible classes, as well as utility functions for defining tasks using these classes, in a way that integrates nicely with Domino Jobs. Notably, the flytekitplugins.domino library provides groups of utilities in the helpers, task, and artifact submodules.

Class constructors

The following example defines a task using the base classes.

from typing import TypeVar, NamedTuple
from flytekit import workflow
from flytekitplugins.domino.task import DominoJobConfig, DominoJobTask

@workflow
def training_workflow(data_path: str) -> FlyteFile[TypeVar("pt")]:

    # Defining a task using Domino classes
    data_prep_job_config = DominoJobConfig(Command="python prep-data.py")
    data_prep_job = DominoJobTask(
        name='Prepare Data',
        domino_job_config=data_prep_job_config,
        inputs={'data_path': str},
        outputs={'processed_data': FlyteFile[TypeVar("csv")]},
        use_latest=True
    )
    data_prep_results = data_prep_job(data_path=data_path)

    return

To pin a specific branch, commit, or tag for one or more of the project’s imported Git repos on a per-task basis, pass ImportedGitRepoOverrides to DominoJobConfig (Git-based projects only):

training_job_config = DominoJobConfig(
    Command="python train-model.py",
    MainRepoGitRef=GitRef(Type="branches", Value="main"),
    ImportedGitRepoOverrides=[
        ImportedGitRepoOverride(
            Name="shared-utils",
            Ref=GitRef(Type="commitId", Value="2f1cb9bf696921f0858"),
        ),
    ],
)

`DominoJobConfig()`

The DominoJobConfig defines the configuration for the Domino Job that is triggered by the task.

Show Parameters

String

The title that is given to the Domino Job. Ignored when calculating the config’s cache version.Example: "My training job"

String

required

The command that is executed in the Domino Job.Example: "python train.py"

String

For projects hosted in the Domino File System, this refers to the commit ID of the code. For Git-based projects, this refers to the commit ID of Artifacts.Example: "953f66f3153b71658d"

GitRef

Reference a specific branch, commit, or tag (for Git-based projects only). See the API guide for more details.Example: GitRef(Type="commitId", Value="2f1cb9bf696921f0858")

String

The resolved commit hash to use. In most cases, if a specific commit is needed, it is best to use the Value argument to GitRef. Specifying the resolved commit explicitly can be used to track a floating reference, but resolve to an older commit.Example: 2f1cb9bf696921f0858

Boolean

default:"False"

When True, floating git references specified in MainRepoGitRef and ImportedGitRepoOverrides are resolved to the latest commit when the Flow is run or relaunched. When False, floating git references resolve the reference on the first run, which is reused when the Flow is relaunched.Example: True

String

The ID of the Domino Hardware Tier. Note that this is different than the name of the hardware tier. Ignored when calculating the config’s cache version.Example: small-k8s

String

The ID of the Domino Environment. Note that this is different than the name or revisionId of the Environment.Example: 6646530dcbd87f1a3dec0050

EnvironmentRevisionSpecification

The revision of the specified Domino Environment. See the EnvironmentRevisionSpec example after the table.

Optional[ClusterProperties]

Compute cluster properties for the job. See the ComputeClusterProperties example after the table.

Float

The amount of disk space (in GiB) to allocate. Ignored when calculating the config’s cache version.Example: 10.0

List[DatasetSnapshot]

List of the dataset snapshots to include in the job. Note that version 0 of a dataset cannot be used, since it is mutable. You must take a snapshot first before using a dataset in a flow.Example: [DatasetSnapshot(Id="6615af2820a4", Version=1)]

List[NetAppVolumeSnapshot]

List of the NetApp Volume snapshots to include in the job. Note that version 0 of a NetAppVolume cannot be used, since it is mutable. You must take a snapshot first before using a NetApp Volume in a flow.Example: [NetAppVolumeSnapshot(Id="5d1204c2-24c8", Version=1)]

List[String]

List of the external data volume mounts (referenced by ID) to include in the Job.Example: ["9625af24kida4dc035aa881b7"]

List[ImportedGitRepoOverride]

Pin a specific branch, commit, or tag for one or more of the project’s imported Git repos, per task (for Git-based projects only). Name must match an imported repository configured on the project, and Ref is a GitRef.Example: [ImportedGitRepoOverride(Name="shared-utils", Ref=GitRef(Type="branches", Value="main"))]

Optional[Tuple[str, ...]]

default:"None"

Optional tuple of names of arguments to the DominoJobConfig to ignore when calculating the config’s cache version.Example: ("EnvironmentId",)

`DominoJobTask()`

Calling the Domino Job task with the relevant inputs (data_prep_job(data_path=data_path)) runs the Domino Job and returns the results as a Promise, which can be used as an input to downstream tasks.

Show Parameters

String

required

The name that is given to the task.Example: My training task

DominoJobConfig

required

The job configuration, as defined above.Example: DominoJobConfig(Command="python prep-data.py")

Dict[String, Type]

Inputs that are required by the task. See above for different input types that are supported. Inputs may be specified from the workflow OR may be outputs from other tasks.Example: {'data_path': str}

Dict[String, Type]

Outputs that are produced by the task. See above for different output types that are supported.Example: {'processed_data': FlyteFile[TypeVar("csv")]}

Boolean

If set to True, this uses the latest project defaults for parameters that were not explicitly provided, like the compute environment version and hardware tier. For better reproducibility, it is recommended to set this to False and explicitly define the necessary parameters.Example: False

Union[bool, Cache]

default:"False"

Indicates if caching should be enabled. Can be a bool or a Flyte Cache object. Setting cache=true is a simple way to enable caching for a task, but using a Cache object may be more flexible depending on your use case. See the caching documentation for a full explanation of how caching works.Example: True

String

When cache=true, the cache version to use. Changes to the task signature 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.Example: "1.0"

Tuple[str, ...]

When cache=true, variable names of inputs to the DominoJobTask that should not be included when calculating the hash used for caching. If not provided, all input variables are included when calculating the hash.Example: (batch_size,)

Integer

Number of times to retry this task during a workflow execution. This can be used to help automatically mitigate intermittent failures.Example: 0

Union[timedelta, int]

The maximum amount of time for which one execution of this task should be executed. The execution is terminated if the runtime exceeds the given timeout.Example: timedelta(hours=3)

`EnvironmentRevisionSpec()`

EnvironmentRevisionSpecification(
    EnvironmentRevisionType=EnvironmentRevisionType.SomeRevision,
    EnvironmentRevisionId="6659daf5fc8de",
)

`ComputeClusterProperties()`

ClusterProperties(
    ClusterType=ComputeClusterType.Dask,
    ComputeEnvironmentId="computeEnvId",
    WorkerHardwareTierId="workerHardwareTierId",
    WorkerCount=3,
    WorkerStorageGiB=100.0,
    MaxWorkerCount=10,
    ComputeEnvironmentRevisionSpec=EnvironmentRevisionSpecification(
        EnvironmentRevisionType.SomeRevision,
        "revisionId",
    ),
    MasterHardwareTierId="masterHardwareTierId",
    ExtraConfigs={"foo": "bar"},
)

Helpers

`run_domino_job_task()`

Helper methods reduce the amount of code necessary to invoke a task. Instead of separately defining a DominoJobConfig and passing it to a DominoJobTask in the examples above, use run_domino_job_task to define the task config, job, inputs and outputs all using a single function call.

To pin imported Git repos for a task, use the base DominoJobConfig and DominoJobTask classes with ImportedGitRepoOverrides. The run_domino_job_task helper also accepts imported_git_repo_overrides.

@workflow
def training_workflow(data_path: str) -> FlyteFile[TypeVar("pt")]:

    # First task using helper method
    data_prep_results = run_domino_job_task(
        flyte_task_name="Prepare data",
        command="python prep-data.py",
        inputs=[Input(name="data_path", type=str, value=data_path)],
        output_spec=[Output(name="processed_data", type=FlyteFile[TypeVar("csv")])],
        use_project_defaults_for_omitted=True
    )

    # Output from the task above will be used in the next step

    return # Outputs will be returned here

Show Parameters

String

The title that is given to the task.Example: My training task

String

The title that is given to the Domino Job.Example: My training job

Boolean

If set to True, this uses the latest project defaults for parameters that were not explicitly provided. For better reproducibility, it is recommended to set this to False and explicitly define the necessary parameters.Example: False

String

For projects hosted in the Domino File System, this refers to the commit ID of the code. For Git-based projects, this refers to the commit ID of the artifacts.Example: 953f66f3153b71658

GitRef

Reference a specific branch, commit, or tag (for Git-based projects only). See the API guide for more details.Example: GitRef(Type="commitId", Value="2f1cb9bf696921f0")

String

Name of the Environment to use in the Domino Job.Example: Domino Standard Environment

String

ID of the Environment to use in the Domino Job. This is recommended over using environment_name to prevent breaking reproducibility when environment names get changed.Example: 6646530dcbd

String

A specific revisionId of the environment to use.Example: 6659daf5fc8

String

Name of the hardware tier to use in the Domino Job.Example: Large

String

ID of the hardware tier to use in the Domino Job. This is recommended over using hardware_tier_name to prevent breaking reproducibility when hardware tier names get changed.Example: small-k8s

List[Input]

Inputs that are required by the task. See above for different input types that are supported. See Input() for details.

List[Output]

Outputs that are produced by the task. See above for different output types that are supported.Example: [Output(name="processed_data", type=FlyteFile[TypeVar("csv")])]

Integer

The amount of disk space (in GiB) to allocate.Example: 10

List[DatasetSnapshot]

List of the dataset snapshots to include in the Job. Note that version 0 of a dataset cannot be used, since it is mutable. You must first take a snapshot before using a dataset in a flow.Example: [DatasetSnapshot(Id="6615af2820a4", Version=1)]

List[NetAppVolumeSnapshot]

List of the NetApp Volume snapshots to include in the Job. Note that version 0 of a NetApp Volume cannot be used, since it is mutable. You must first take a snapshot before using a NetApp Volume in a flow.Example: [NetAppVolumeSnapshot(Id="5d1204c2-24c8", Version=1)]

List[str]

List of the external data volume mounts (referenced by ID) to include in the Job.Example: [9625af24kida4dc035aa881b7]

Union[bool, Cache]

Indicates if caching should be enabled. Can be a bool or a Flyte Cache object. Defaults to false. Setting cache=true is a simple way to enable caching for a task, but using a Cache object may be more flexible depending on your use case. See the caching documentation for a full explanation of how caching works.Example: False

String

Tuple[str, ...]

Integer

Number of times to retry this task during a workflow execution.Example: 0

Union[timedelta, int]

The maximum amount of time for which one execution of this task should be executed. The execution is terminated if the runtime exceeds the given timeout.Example: timedelta(hours=3)

`Input()`

Specify a Flows Task input. Inputs include values and files, and must be declared for each Task.

Input(name="number", type=int, value=13)

Input(
    name="data",
    type=FlyteFile[TypeVar("csv")],
    path="/mnt/data/numbers/data.csv"
)

Show Parameters

String

required

The name of the input. When used in a DominoJobTask, this name appears in the Flyte UI as the name of an input to the task.

Type

required

The programmatic type of the input. When used in a DominoJobTask, this type appears in the Flyte UI as the type of an input to the task.

Any

The value of the input. When used in a DominoJobTask, this value is passed to the task for the associated input name, and the value appears in the Flyte UI. The value must be of the type defined by the type parameter.

String

default:"None"

The path to the input file. When used in a DominoJobTask, this path is used to download the file to the local disk. If not provided, the file is downloaded to the default location in /workflow/inputs.

For consistency with workspace execution, paths can be passed to the Dataset (/mnt/data) or NetAppVolume (/mnt/netapp-volumes/) data locations. Data in these locations can be both an input and an output, populating the files when the Task starts and propagating the files as outputs for downstream tasks.

Example: "/mnt/netapp-volumes/project-data/data.csv"

`Output()`

Specify a Flows Task output.

Output(name="number", type=int)

Output(
    name="data",
    type=FlyteFile[TypeVar("csv")],
    path="/mnt/data/numbers/data.csv"
)

Show Parameters

String

required

The name of the input. When used in a DominoJobTask, this name appears in the Flyte UI as the name of an input to the task.

Type

required

The programmatic type of the input. When used in a DominoJobTask, this type appears in the Flyte UI as the type of an input to the task.

String

default:"None"

Example: "/mnt/netapp-volumes/project-data/data.csv"

Next steps

Once you have properly defined the flow, learn how to:

Core Concepts

Features

Flow files

Workflow basics

Inputs & outputs

Flow tasks

Declare tasks

Pass data

Return outputs

Write task code

Read inputs

Write outputs

Domino Flyte plugin

Class constructors

`DominoJobConfig()`

`DominoJobTask()`

`EnvironmentRevisionSpec()`

`ComputeClusterProperties()`

Helpers

`run_domino_job_task()`

`Input()`

`Output()`

Next steps

​Flow files

​Workflow basics

​Inputs & outputs

​Flow tasks

​Declare tasks

​Pass data

​Return outputs

​Write task code

​Read inputs

​Write outputs

​Domino Flyte plugin

​Class constructors

​DominoJobConfig()

​DominoJobTask()

​EnvironmentRevisionSpec()

​ComputeClusterProperties()

​Helpers

​run_domino_job_task()

​Input()

​Output()

​Next steps

Flow files

Workflow basics

Inputs & outputs

Flow tasks

Declare tasks

Pass data

Return outputs

Write task code

Read inputs

Write outputs

Domino Flyte plugin

Class constructors

`DominoJobConfig()`

`DominoJobTask()`

`EnvironmentRevisionSpec()`

`ComputeClusterProperties()`

Helpers

`run_domino_job_task()`

`Input()`

`Output()`

Next steps