I did not realize that the KV store is only supported for Prefect Cloud. Are there any solutions to backfill/differential dataload/watermarking for those of us who are hosting our own Prefect servers?

This user did it with Redis and the Redis tasks from the task library I think if you want to explore that

Forgive my basic questions, but would it be possible to access the a redis client (or similar) through the Prefect context? I guess what I am asking is - can I add custom objects to the context

Hi @Kevin Kho - Once again, thank you for your help. We managed to implement watermarking using Redis as you suggested, however we chose to create a form of the

@task

decorator instead, which allows for watermarked inputs:

@watermarked_task(watermark_args=["count"])
def say_hello_count(count, store=None):
    logger = prefect.context.get("logger")

    watermark = int(count.watermark or 0)
    <http://logger.info|logger.info>("Hello watermark. watermark={}, count={}".format(watermark, count))
    
    value = count.value
    do_somewith_with_value(value)

    store.set(count.key, watermark + 1)

Here, the

count

and

store

arguments are amended by the decorator.

count

is wrapped into a

Watermark

-type which is just a

namedtuple

to hold the

watermark

,

value

, and

key

. The

store

, set to

None

by default to keep the IDE happy, is set to the watermark-store client. We kind of dislike changing inputs this way, but this works fine for now. On the back end, we have created a watermark-store factory-type which will initialize a watermark storage client based on environment variables (or an optional input to the decorator mentioned above). That way, we can allow local watermark management for developers in the form of in-memory stores or files (also as you suggested), while any deployed flow will use Redis. We had to create a separate prefect agent deployment and disable the one in the helm-chart, as the helm-chart does not allow us to set environment variables in the agent, as well as a custom config-map in order to provide a prefect

job_spec.yaml