Uploading and Downloading Data
When moving files and directories between local disks and object storage (such as S3, GCS, or Azure Blob Storage), flyte-sdk uses File[T] and Dir[T] from flyte.io. Unlike legacy systems where type transformers implicitly download inputs or upload outputs behind the scenes, flyte-sdk requires tasks to explicitly perform data transfer.
Uploading Local Data to Remote Storage
To transfer local files or directory trees to remote storage, use the asynchronous from_local class methods on File and Dir.
from pathlib import Path
import pandas as pd
from flyte import task
from flyte.io import File, Dir
@task
async def stage_datasets() -> tuple[File[pd.DataFrame], Dir]:
# 1. Prepare local files and folders
local_file_path = Path("/tmp/metrics.csv")
local_file_path.write_text("metric,value\naccuracy,0.95\n")
local_dir_path = Path("/tmp/models")
local_dir_path.mkdir(parents=True, exist_ok=True)
(local_dir_path / "config.json").write_text('{"epochs": 10}')
(local_dir_path / "weights.bin").write_bytes(b"\x00\x01\x02\x03")
# 2. Upload file to an explicit remote location
remote_file = await File[pd.DataFrame].from_local(
local_path=local_file_path,
remote_destination="s3://my-bucket/experiments/metrics.csv",
)
# 3. Upload directory recursively to raw storage
# Leaving remote_path as None auto-generates a remote URI
remote_dir = await Dir.from_local(
local_path=local_dir_path,
remote_path="s3://my-bucket/experiments/models/",
)
return remote_file, remote_dir
Automatic Path Generation
If remote_destination in File.from_local() or remote_path in Dir.from_local() is omitted (None), flyte-sdk generates a remote URI automatically under the configured raw data storage prefix:
# Uploads to an auto-generated path in configured storage:
remote_file = await File[pd.DataFrame].from_local("/tmp/metrics.csv")
print(remote_file.path) # e.g., 's3://my-configured-bucket/raw/...'
Task Caching with Hash Methods
File.from_local() accepts a hash_method argument to compute a cache key for tasks using this file as an input. You can pass a string literal, a precomputed key, or a streaming hashing accumulator:
from flyte.io import File, HashlibAccumulator, PrecomputedValue
# 1. Compute hash automatically during file upload using sha256
remote_file = await File.from_local(
local_path="/tmp/metrics.csv",
remote_destination="s3://my-bucket/metrics.csv",
hash_method=HashlibAccumulator.from_hash_name("sha256"),
)
# 2. Pass a known precomputed hash
precomputed_file = await File.from_local(
local_path="/tmp/metrics.csv",
remote_destination="s3://my-bucket/metrics.csv",
hash_method=PrecomputedValue("a1b2c3d4e5f6"),
)
For directories, pass dir_cache_key directly to Dir.from_local():
remote_dir = await Dir.from_local(
local_path="/tmp/models/",
remote_path="s3://my-bucket/models/",
dir_cache_key="model-run-v1-hash",
)
Streaming Directly to Remote Storage
When writing large files from within a task, streaming directly to blob storage avoids writing intermediate data to local disk. Call File.new_remote() to allocate a destination path in raw storage, then open the file handle with binary write mode ("wb"):
import pandas as pd
from flyte import task
from flyte.io import File
@task
async def stream_large_dataset() -> File[pd.DataFrame]:
df = pd.DataFrame({"id": range(1000), "score": [0.1 * i for i in range(1000)]})
# Allocates a unique remote destination URI in raw storage
remote_file = File.new_remote()
# Stream dataframe directly to cloud storage in binary mode
async with remote_file.open("wb") as stream:
df.to_csv(stream, index=False)
return remote_file
Downloading Remote Data to Local Storage
Use download() on File or Dir instances to pull remote objects down to the local filesystem.
from pathlib import Path
import pandas as pd
from flyte import task
from flyte.io import File, Dir
@task
async def process_data(data_file: File[pd.DataFrame], model_dir: Dir):
# Download file to a specific local path
local_file_path = await data_file.download("/tmp/downloaded_data.csv")
# Download directory to an auto-generated local temp path
local_dir_path = await model_dir.download()
print(f"File downloaded to: {local_file_path}")
print(f"Directory tree downloaded to: {local_dir_path}")
# Inspect downloaded local directory contents
for path in Path(local_dir_path).iterdir():
print(f"Found local file: {path.name}")
If local_path is omitted, flyte-sdk generates a path in the local temporary directory via storage.get_random_local_path().
Streaming Remote Data Without Full Downloads
When working with large remote files or directory trees, you can read files directly without saving them to local disk first.
Streaming File Contents
Use async with file.open("rb") to read file streams directly from object storage:
import pandas as pd
from flyte import task
from flyte.io import File
@task
async def parse_remote_csv(csv_file: File[pd.DataFrame]) -> int:
# Stream directly from S3/GCS/ABFS without downloading to disk
async with csv_file.open("rb") as stream:
df = pd.read_csv(stream)
return len(df)
For synchronous execution contexts, use open_sync("rb"):
with csv_file.open_sync("rb") as stream:
content = stream.read()
Traversing Remote Directories
To inspect or process files inside a remote Dir without downloading the full tree, use walk() or list_files():
from flyte import task
from flyte.io import Dir
@task
async def inspect_directory(remote_dir: Dir):
# 1. Traverse all nested files asynchronously
async for file in remote_dir.walk(recursive=True):
print(f"Found remote file: {file.path}")
async with file.open("rb") as f:
header = await f.read(100)
print(f"File header: {header}")
# 2. Get a flat list of top-level files (non-recursive)
top_level_files = await remote_dir.list_files()
print(f"Total top-level files: {len(top_level_files)}")
# 3. Retrieve a specific file by name
config_file = await remote_dir.get_file("config.json")
if config_file:
async with config_file.open("rb") as f:
config_data = await f.read()
Synchronous alternatives remote_dir.walk_sync(), remote_dir.list_files_sync(), and remote_dir.get_file_sync(file_name) are also available.
Referencing Existing Remote Storage
If data already exists in cloud object storage, reference it without initiating a network transfer:
from flyte.io import File, Dir
# Reference existing remote file
remote_file = File.from_existing_remote(
remote_path="s3://my-bucket/existing_dataset.parquet",
file_cache_key="dataset-v1-hash", # Optional cache key
)
# Reference existing remote directory
remote_dir = Dir.from_existing_remote(
remote_path="s3://my-bucket/models/v1/",
dir_cache_key="models-v1-hash", # Optional cache key
)
Troubleshooting and Gotchas
ValueError: Mode must include 'b' for binary access
Remote storage backends accessed through File.open() require binary modes ("rb" or "wb"). Text modes like "r" or "w" raise a ValueError when opened on remote URIs.
# Incorrect:
# async with remote_file.open("r") as f: ...
# Correct:
async with remote_file.open("rb") as f:
raw_data = await f.read()
text = raw_data.decode("utf-8")
NotImplementedError with Sync Operations on Remote Paths
Methods such as Dir.from_local_sync() and Dir.download_sync() raise NotImplementedError when called on remote paths. When moving data to or from object storage, use the async methods: await Dir.from_local(...) and await Dir.download(...).
InitializationError on from_local or new_remote
Methods that resolve remote storage paths (File.from_local, File.new_remote) depend on context from flyte.init(). If running standalone scripts or uninitialized tests outside tasks, ensure the Flyte runtime is initialized before calling these methods.
Local Path Pass-Through Optimization
If File.from_local(local_path) is called without remote_destination and the configured storage location is the local filesystem (file://), flyte-sdk does not duplicate the file. It points directly to the absolute path of the existing local file.