What is necessary to get Prefect to use checkpointing? It seems there is something missing from my simplified test code, but unclear what:

from prefect import Flow, task, unmapped, Parameter
from prefect.engine.results import LocalResult
from prefect.engine.executors import LocalDaskExecutor
import prefect

lr = LocalResult(location='{flow_name}-{task_name}-{map_index}.txt')

@task(log_stdout=True, checkpoint=True,)
def add(x, y):
    return x + y

with Flow('iterated map', result=lr) as flow:
    y = unmapped(Parameter('y', default=10))
    mapped_result = add.map([1, 2, 3], y=y)

flow.run(executor=LocalDaskExecutor())

Hi jorwoods - to turn on checkpointing locally you need to set

export PREFECT__FLOWS__CHECKPOINTING=true

Ahh. For some reason I misread it as needing either that environment variable OR the kwarg, not both.