Skip to main content

Local Task Execution with Docker

Run a container task locally with Docker​

By the end of this tutorial, you will have a flyte.extras.ContainerTask that accepts a scalar input, runs a Docker image on your machine, writes a named output file through a mounted output directory, streams container logs, and returns the output as a Python value.

Prerequisites​

You need:

  • flyte-sdk, with the Docker SDK for Python installed separately:

    pip install docker

  • A running, reachable Docker engine. ContainerTask.execute creates its client with docker.from_env(), so it uses the Docker SDK's standard environment rather than a Flyte-specific configuration key.

  • A container image that the Docker daemon can pull, or that is already available locally.

The Docker Python package is imported only when execute runs. If it is missing, execution raises an ImportError with the installation command above.

1. Define a raw-container task​

Import ContainerTask from the public flyte.extras package and declare the input and output types in its constructor:

from flyte.extras import ContainerTask

task = ContainerTask(
    name="write-result",
    image="python:3.11-slim",
    command=[
        "python",
        "-c",
        "import pathlib, sys; pathlib.Path('/var/outputs/result').write_text(sys.argv[1])",
        "{{.inputs.text}}",
    ],
    inputs={"text": str},
    outputs={"result": str},
)

flyte.extras re-exports the concrete class implemented in flyte.extras._container. The constructor records a raw-container task, creates a NativeInterface from inputs and outputs, and accepts these execution settings:

  • name identifies the task.
  • image is a string image reference or an Image object.
  • command is implemented as a list of strings. arguments, when supplied, are appended to that list.
  • inputs and outputs map names to Python types.
  • input_data_dir defaults to /var/inputs.
  • output_data_dir defaults to /var/outputs.
  • metadata_format defaults to "JSON" and accepts "JSON", "YAML", or "PROTO".
  • local_logs defaults to True.

Although the class docstring mentions a single-string command, the implementation concatenates the stored command and arguments as lists. Pass a list for command and, if used, a list for arguments.

2. Understand image resolution and pulling​

String images are converted during construction. The special string "auto" selects Image.from_debian_base(); any other string is passed to Image.from_base. You can also provide an image object directly:

from flyte import Image
from flyte.extras import ContainerTask

image = Image.from_base("python:3.11-slim")
task = ContainerTask(
    name="write-result",
    image=image,
    command=[
        "python",
        "-c",
        "import pathlib, sys; pathlib.Path('/var/outputs/result').write_text(sys.argv[1])",
        "{{.inputs.text}}",
    ],
    inputs={"text": str},
    outputs={"result": str},
)

When execution starts, the task reads the resulting image object's uri. It asks Docker for images matching that URI with client.images.list(filters={"reference": image}). Only when that list is empty does it call client.images.pull(image). Pull failures are logged and re-raised. Registry authentication, daemon connectivity, and image compatibility therefore remain Docker-level concerns.

3. Resolve scalar inputs in the command​

The {{.inputs.<name>}} form is handled by _render_command_and_volume_binding. For a scalar input, the matching value is converted with str() and substituted into the command. The task above therefore produces a command whose final argument is the value passed as text.

Execute the task asynchronously:

result = await task.execute(text="hello from Docker")
print(result)

The local execution path prints the prepared command unconditionally. With the default local_logs=True, it also prints each streamed Docker log as [Local Container] .... The returned value is a tuple because _get_output returns one tuple element per declared output, so the visible result is:

('hello from Docker',)

4. Mount File and Dir inputs by container path​

Scalar template substitution is separate from file and directory mounting. For flyte.io.File and flyte.io.Dir, use a command path beginning with the configured input_data_dir:

from flyte.extras import ContainerTask
from flyte.io import File

file_task = ContainerTask(
    name="read-input-file",
    image="python:3.11-slim",
    command=[
        "python",
        "-c",
        "import pathlib; pathlib.Path('/var/outputs/size').write_text(str(pathlib.Path('/var/inputs/infile').stat().st_size))",
        "/var/inputs/infile",
    ],
    inputs={"infile": File},
    outputs={"size": int},
)

For an input whose value is exactly a File or Dir, the implementation extracts infile from /var/inputs/infile, then adds a Docker volume binding from input_val.path on the host to /var/inputs/infile in the container. The binding uses {"mode": "rw"}. The same convention applies to a directory input, for example /var/inputs/input_dir with an input named input_dir and type Dir.

Do not write {{.inputs.infile}} for a File or Dir. That form raises an AssertionError instructing you to use path-like syntax. The path extractor is based on the configured input directory and recognizes names made from word characters, hyphens, and periods; nested or unusual paths may not be recognized as input keys. The special mount check is also an exact type check, so subclasses do not take this branch.

5. Write outputs through the mounted output directory​

execute obtains a host-side directory from storage.get_random_local_directory() and adds this binding:

volume_bindings[str(output_directory)] = {
    "bind": self._output_data_dir,
    "mode": "rw",
}

With the defaults, the container sees that directory at /var/outputs. For every declared output key, write a file or directory named with that key. The earlier task writes /var/outputs/result, matching outputs={"result": str}.

After starting the container detached with remove=True, execute streams logs when enabled, calls container.wait(), and reads the output directory. _get_output looks for <temporary-output-directory>/<output key>. Existing files are read completely as text; missing paths produce None. Values are converted in the insertion order of the output dictionary and returned as a tuple.

The conversion rules include:

  • bool: any value whose lowercase text is not exactly "false" becomes True.
  • datetime.datetime: parsed with datetime.datetime.fromisoformat.
  • datetime.timedelta: parsed by _string_to_timedelta.
  • flyte.io.File: loaded with await File.from_local(output_path).
  • flyte.io.Dir: loaded with await Dir.from_local(output_path).
  • Other types: constructed by calling the declared type with the output text.

For example, the output declaration can contain several supported conversions:

import datetime
from flyte.extras import ContainerTask
from flyte.io import Dir, File

typed_task = ContainerTask(
    name="typed-results",
    image="python:3.11-slim",
    command=["/bin/sh", "-c", "..."],
    outputs={
        "ok": bool,
        "when": datetime.datetime,
        "elapsed": datetime.timedelta,
        "artifact": File,
        "folder": Dir,
    },
)

The container must create /var/outputs/ok, /var/outputs/when, /var/outputs/elapsed, /var/outputs/artifact, and /var/outputs/folder for that declaration. File.from_local and Dir.from_local are asynchronous and can require Flyte storage initialization and configured remote storage even though the container itself runs locally. Missing output files can consequently fail during conversion—for example, bool conversion calls .lower() on the value.

6. See the serialized task contract​

ContainerTask also supplies the task-template extension points used by Flyte serialization. container_args returns the command followed by arguments. data_loading_config enables data loading, serializes the input and output container paths, and maps the three metadata formats to the corresponding tasks_pb2.DataLoadingConfig enum.

The runtime serializer consumes both methods when it constructs the protobuf container:

return tasks_pb2.Container(
    image=img_uri,
    command=[],
    args=task_template.container_args(serialize_context),
    resources=resources,
    env=env,
    data_config=task_template.data_loading_config(serialize_context),
    config=task_template.config(serialize_context),
)

Thus, the paths used for local volume mounting are also exposed as the raw-container data-loading paths during serialization. If metadata_format is one of "JSON", "YAML", or "PROTO", it is mapped to the matching protobuf value; the implementation uses a JSON fallback if an unsupported value reaches the method.

Complete local result​

This is the minimal scalar-input flow in one place:

from flyte.extras import ContainerTask

task = ContainerTask(
    name="write-result",
    image="python:3.11-slim",
    command=[
        "python",
        "-c",
        "import pathlib, sys; pathlib.Path('/var/outputs/result').write_text(sys.argv[1])",
        "{{.inputs.text}}",
    ],
    inputs={"text": str},
    outputs={"result": str},
    local_logs=True,
)

result = await task.execute(text="hello from Docker")
print(result)

With Docker available, the image is pulled if necessary, the command is printed, the container's logs are streamed, /var/outputs is mounted to a temporary host directory, and the output is returned as ('hello from Docker',).

Troubleshoot the local run​

  • Docker package or daemon: Install docker with pip install docker, then ensure docker.from_env() can reach the local engine.
  • Image pull: A pull occurs only when no matching local image is listed. Pull and registry errors are re-raised after logging.
  • File or directory input: Use /var/inputs/<input-name> (or the configured input_data_dir) in the command, not {{.inputs.<input-name>}}.
  • No output: Write the output under the configured output_data_dir using the declared output key. Missing paths are passed to the output converter as None.
  • Logs: local_logs=False suppresses streamed container logs, but the prepared command is still printed.
  • Hung or failing container: container.wait() is called without a timeout and its status is not inspected. The source contains a TODO for a timeout, so a hung container can wait indefinitely and a non-zero exit status is not explicitly converted into a Python exception.

As next steps, use File or Dir path bindings for host-side input data, declare typed outputs when your container writes the corresponding output files, and configure Flyte storage before using File or Dir outputs.