Automated Data Movement in Workflows

Flyte Copilot automates the movement of data between remote storage (like S3 or GCS) and the local filesystem where a task executes. By acting as an "init" container for downloads and a "sidecar" container for uploads, it allows user code to interact with local files instead of cloud-specific SDKs.

Downloading Task Inputs

When a task starts, the Downloader retrieves the input metadata and materializes the actual data onto the local disk. This process ensures that by the time the user's container runs, all required files and primitive values are available in a predictable directory structure.

Using the Downloader

You typically invoke the downloader within a command-line tool or an init container. The DownloadInputs method is the primary entry point:

// Example based on flytecopilot/cmd/download.go
dl := data.NewDownloader(ctx, dataStore, core.DataLoadingConfig_JSON, core.IOStrategy_DOWNLOAD_EAGER)

err := dl.DownloadInputs(
    ctx, 
    "s3://my-bucket/inputs.pb", // Remote path to input LiteralMap
    "/var/lib/flyte/inputs",    // Local directory to store data
)

Internal Mechanism

The Downloader (defined in flytecopilot/data/download.go) performs the following steps:

1. Metadata Retrieval: It reads the LiteralMap from the remote inputRef using the provided DataStore.
2. Recursive Materialization: The RecursiveDownload method iterates through every variable in the input map. If a literal contains OffloadedMetadata, the downloader first resolves the actual data from the offloaded location.
3. Concurrency: It uses flytestdlib/futures to download multiple inputs in parallel. Each variable is processed in its own goroutine.
4. Blob Handling:
   • Single Blobs: Downloaded directly to the target file path.
   • Multipart Blobs: The downloader lists all parts of the remote prefix. It uses a batch size of 100 (maxItems := 100) when calling the storage List API. It preserves the relative directory structure of the remote parts when writing them locally.
5. Local Artifacts: After downloading, it writes an inputs.pb file to the local directory. If configured for JSON or YAML, it also generates inputs.json or inputs.yaml containing the primitive values and local paths to blobs.

Uploading Task Outputs

Once a task completes, the Uploader scans the local output directory and transfers files back to remote storage based on the task's output interface.

Using the Uploader

The uploader requires a VariableMap (the "interface") to know how to interpret local files—for example, whether a file should be uploaded as a Blob or read as a Simple string.

// Example based on flytecopilot/cmd/sidecar.go
ul := data.NewUploader(ctx, dataStore, core.DataLoadingConfig_JSON, core.IOStrategy_UPLOAD_ON_EXIT, "_ERROR")

err := ul.RecursiveUpload(
    ctx, 
    outputInterface,           // *core.VariableMap defining expected outputs
    "/var/lib/flyte/outputs",  // Local directory containing results
    "s3://my-bucket/outputs.pb", // Where to write the final metadata
    "s3://my-bucket/raw/",       // Prefix for raw data (blobs)
)

Type Mapping and Constraints

The Uploader (defined in flytecopilot/data/upload.go) maps local filesystem entities to Flyte types:

• Simple Types: For types like Integer, String, or Boolean, the uploader reads the content of a file named after the variable.
  • Constraint: Simple types are subject to a maxPrimitiveSize of 1024 bytes. If a file intended to be a primitive exceeds this limit, the upload fails.
• Blobs:
  • If the local path is a file, it is uploaded as a single blob.
  • If the local path is a directory, the uploader performs a filepath.Walk and uploads every file within it, creating a multipart blob in remote storage.

Error Reporting via Files

Flyte Copilot provides a standardized way for tasks to report failures by writing to a specific error file (usually named _ERROR).

In RecursiveUpload, the first action the Uploader takes is checking for this file:

// flytecopilot/data/upload.go
errFile := path.Join(fromPath, u.errorFileName)
if info, err := os.Stat(errFile); err == nil {
    // ... validation ...
    b, err := os.ReadFile(errFile)
    return errors.Errorf("User Error: %s", string(b))
}

If the error file exists:

1. The uploader reads the content (up to 1MB).
2. It stops the upload process immediately.
3. It returns a "User Error" containing the file's content, which Flyte uses to provide feedback in the UI.

Configuration and Formats

The behavior of data movement is controlled by several parameters passed during instantiation:

Parameter	Description
format	Determines the format of the local metadata file (JSON, YAML, or PROTO).
mode	For downloads, EAGER (default) or LAZY. For uploads, ON_EXIT or ON_WRITE.
errorFileName	The name of the file the uploader should check for task-reported errors (e.g., _ERROR).

The Unmarshal type is also used throughout these processes to define how protobuf messages are deserialized from various formats:

type Unmarshal func(r io.Reader, msg proto.Message) error