Job scheduling and hibernation

With job hibernation it is possible to spread the execution of a job out over time. This can be especially useful if depending on outside inputs, such as third party APIs, or if the job is reading from a slowly updating data source. In these cases, instead of a job idling while waiting for input (and costing credits!), the job can hibernate, essentially shutting down the process, and thus not using any resources.

The job can be manually resumed again, but if the job is configured with scheduling, then it will be automatically resumed at the next scheduled resume time.

However, in order for these features to be especially useful, the job needs to either be stateless, or use the state to store the information required to continue after being resumed.

Managing job state

You can get and set values in the state, as shown below. When using get() to access an unset state value, None is returned, unless the default argument is set.

Example: Get and set simple values in the state

with JobClient.connect() as job:
    my_value = job.state.get("my_stored_value", default=0.0)
    my_value += 20.0
    job.state.set("my_stored_value", default=my_value)

    my_optional_value = job.state.get("my_optional_stored_value")
    if my_optional_value is None:
        my_optional_value = 10.0
    my_optional_value += 10.0
    job.state.set("my_optional_stored_value", my_optional_value)

The state values have to be JSON-(de)serializable, so complex objects are not currently directly supported to be stored, however if you can serialize them into basic python types, and deserialize them yourself, then it is possible to store anything you need. In the example below, we show how more advanced serialization could work.

Example: Serialize and deserialize custom objects

class MySerializableObject:
    def __init__(self, my_value: float, my_time: datetime) -> None:
        self.my_value = my_value
        self.my_time = my_time

    def serialize(self) -> str:
        return json.dumps(
            {
                "type": "MySerializableObject",
                "my_value": self.my_value,
                "my_time": self.my_time.isoformat(),
            }
        )

    @staticmethod
    def deserialize(serialized: str) -> "MySerializableObject":
        data = json.loads(serialized)
        if data["type"] != "MySerializableObject":
            # This check is just to ensure that we did in fact serialize
            # this data as this kind of object
            raise TypeError(f"Got an unexpected object type {data['type']}")

        return MySerializableObject(
            my_value=data["my_value"],
            my_time=datetime.fromisoformat(data["my_time"]),
        )


with JobClient.connect() as job:
    my_serialized_object = job.state.get("my_stored_serialized_object")
    if my_serialized_object:
        # If some serialized object was saved
        my_deserialized_object = MySerializableObject.deserialize(
            my_serialized_object
        )
    else:
        # else, we create a new object
        my_deserialized_object = MySerializableObject(
            my_value=5.0,
            my_time=datetime.now(),
        )

    # And we modify it
    my_deserialized_object.my_value += 20.0
    my_deserialized_object.my_time += timedelta(days=5)

    # And finally serialize and save it in the state again
    job.state.set("my_stored_serialized_object", my_deserialized_object.serialize())

Hibernating jobs

Now that you can manage state, hibernation starts to be useful, since it is possible to store values across hibernations.

A job can either hibernate by choosing to do so itself with hibernate(), or by being sent a hibernation command, if the user requested the job to hibernate from the platform UI.

A callback function can be specified with set_hibernate_callback() that will be run before the job shuts down, when it is hibernated. It is called for hibernations triggered by the job itself, and for hibernations triggered by commands.

Example: Handle and trigger hibernations

with JobClient.connect() as job:

    def my_hibernation_callback():
        hibernate_count = job.state.get("hibernate_count", 0)
        hibernate_count += 1
        job.state.set("hibernate_count", hibernate_count)

    # Configure the hibernation callback
    job.set_hibernate_callback(my_hibernation_callback)

    # Hibernate the job
    job.hibernate()

Job time across hibernates

When a job that is ranging through time (e.g. using range() or read_inputs()) is hibernated, the current time of the job is stored, and when a job is resumed, it is loaded again.

This means that the two previously mentioned methods will restart from the time of the hibernation, when the job is resumed.

Reading data across hibernates

When a job is hibernated while reading data using read_inputs(), the progress of the reading is stored (as mentioned), and upon resume the job will continue reading from the same point.

Writing data across hibernates

When writing data within a loop of either range() or read_inputs(), the current time is stored upon hibernation.

In the below example, writing will be resumed from the time of hibernation, after the job is resumed, because writing is done within a loop that ranges over the job time.

Example: Continue writing upon resume

with JobClient.connect() as job:
    for t in job.job_time.range(step=timedelta(hours=1))
        job.writer.row(t, {"value": 123.0})