flytekit.core.checkpointer

Directory

Classes

flytekit.core.checkpointer.Checkpoint

Base class for Checkpoint system. Checkpoint system allows reading and writing custom checkpoints from user scripts

Methods

prev_exists()

def prev_exists()

read()

def read()

This should only be used if there is a singular checkpoint file written. If more than one checkpoint file is found, this will raise a ValueError

restore()

def restore(
    path: typing.Union[pathlib.Path, str],
) -> typing.Optional[pathlib.Path]

Given a path, if a previous checkpoint exists, will be downloaded to this path. If download is successful the downloaded path is returned

save()

def save(
    cp: typing.Union[pathlib.Path, str, _io.BufferedReader],
)

Usage: If you have a io.BufferedReader then the following should work

with input_file.open(mode="rb") as b:
    checkpointer.save(b)

write()

def write(
    b: bytes,
)

This will overwrite the checkpoint. It can be retrieved using read or restore

flytekit.core.checkpointer.SyncCheckpoint

This class is NOT THREAD-SAFE! Sync Checkpoint, will synchronously checkpoint a user given file or folder. It will also synchronously download / restore previous checkpoints, when restore is invoked.

TODO: Implement an async checkpoint system

Parameters

class SyncCheckpoint(
    checkpoint_dest: str,
    checkpoint_src: typing.Optional[str],
)

checkpoint_src: If a previous checkpoint should exist, this path should be set to the folder that contains the checkpoint information checkpoint_dest: Location where the new checkpoint should be copied to

Methods

prev_exists()

def prev_exists()

read()

def read()

This should only be used if there is a singular checkpoint file written. If more than one checkpoint file is found, this will raise a ValueError

restore()

def restore(
    path: typing.Union[pathlib.Path, str, NoneType],
) -> typing.Optional[pathlib.Path]

Given a path, if a previous checkpoint exists, will be downloaded to this path. If download is successful the downloaded path is returned

save()

def save(
    cp: typing.Union[pathlib.Path, str, _io.BufferedReader],
)

Usage: If you have a io.BufferedReader then the following should work

with input_file.open(mode="rb") as b:
    checkpointer.save(b)

write()

def write(
    b: bytes,
)

This will overwrite the checkpoint. It can be retrieved using read or restore