Configuring Data Storage

The flyte-sdk storage module provides a unified interface for interacting with cloud storage backends like AWS S3, Google Cloud Storage (GCS), and Azure Blob Storage (ABFS). It leverages the fsspec library and obstore to handle remote file operations while allowing configuration through environment variables or explicit initialization.

Automatic Configuration via Environment Variables

The flyte-sdk automatically detects the storage backend based on the URI scheme (e.g., s3://, gs://, abfs://) and configures the connection using specific environment variables.

AWS S3 and S3-Compatible Storage

For S3, flyte-sdk uses the storage.S3 class. It reads the following environment variables:

Environment Variable	Description
FLYTE_AWS_ENDPOINT	The S3 endpoint URL (e.g., for MinIO).
FLYTE_AWS_ACCESS_KEY_ID	AWS Access Key ID.
FLYTE_AWS_SECRET_ACCESS_KEY	AWS Secret Access Key.

Google Cloud Storage (GCS)

For GCS, flyte-sdk uses the storage.GCS class.

Environment Variable	Description
GCP_GSUTIL_PARALLELISM	Boolean to enable/disable gsutil parallelism.

Azure Blob Storage (ABFS)

For Azure, flyte-sdk uses the storage.ABFS class.

Environment Variable	Description
AZURE_STORAGE_ACCOUNT_NAME	Azure Storage account name.
AZURE_STORAGE_ACCOUNT_KEY	Azure Storage account key.
AZURE_TENANT_ID	Azure tenant ID for service principal auth.
AZURE_CLIENT_ID	Azure client ID for service principal auth.
AZURE_CLIENT_SECRET	Azure client secret for service principal auth.

Global Storage Settings

Common settings for all providers are defined in the base storage.Storage class:

Environment Variable	Default	Description
UNION_STORAGE_RETRIES	3	Number of retries for storage operations.
UNION_STORAGE_BACKOFF_SECONDS	5	Backoff time between retries.
UNION_STORAGE_DEBUG	False	Enable debug logging for storage.

Explicit Configuration

If you prefer not to use environment variables, you can manually instantiate a storage configuration and pass it to flyte.init().

import flyte
from flyte.storage import S3

# Manually configure S3
s3_config = S3(
    endpoint="https://my-custom-s3.com",
    access_key_id="my-key",
    secret_access_key="my-secret",
    retries=5
)

# Initialize flyte-sdk with the custom storage config
flyte.init(storage=s3_config)

Configuring for Local Sandbox (MinIO)

The S3 class provides a helper method for_sandbox() to quickly configure flyte-sdk for a local MinIO instance (typically used in local development environments).

from flyte.storage import S3
import flyte

# Configures for http://localhost:4566 with default minio credentials
sandbox_storage = S3.for_sandbox()
flyte.init(storage=sandbox_storage)

Data Transfer Operations

Once configured, you can use the storage module to move data between local and remote paths.

Uploading and Downloading Files

Use storage.put() to upload and storage.get() to download.

import flyte.storage as storage

# Upload a local file to S3
remote_path = await storage.put(
    from_path="/tmp/local_data.csv", 
    to_path="s3://my-bucket/data/remote_data.csv"
)

# Download a file from GCS to a local path
local_path = await storage.get(
    from_path="gs://my-bucket/results/output.json",
    to_path="/tmp/downloaded_output.json"
)

Streaming Data

For large datasets, use storage.put_stream() and storage.get_stream() to handle data as an async iterable of bytes.

import flyte.storage as storage

async def data_generator():
    yield b"first chunk,"
    yield b"second chunk"

# Stream data to Azure Blob Storage
await storage.put_stream(
    data_iterable=data_generator(),
    to_path="abfs://my-container/streamed_file.txt"
)

# Stream data from S3
async for chunk in storage.get_stream("s3://my-bucket/large_file.bin"):
    process(chunk)

Troubleshooting and Gotchas

Obstore Bypass

The storage module includes a bypass for the obstore library in put_stream and get_stream. This is a workaround for obstore's fsspec implementation which may not support certain async operations natively. If you encounter issues with streaming on S3, GCS, or ABFS, ensure obstore is correctly installed as it is the preferred high-performance backend for these protocols in flyte-sdk.

Path Joining

The storage.join() function is currently a wrapper around os.path.join().

import flyte.storage as storage

# Joins paths using standard OS separators
full_path = storage.join("s3://my-bucket", "folder", "file.txt")

Be aware that this may not handle all remote path edge cases (like trailing slashes on different protocols) as robustly as a full fsspec implementation.

Anonymous Access

If an operation fails due to authentication errors, flyte-sdk will automatically attempt to retry the operation with anonymous=True (skipping signatures) if the underlying provider supports it. This is handled internally within storage.get().