if I return a lot of data serialized with GCS results, is it expected that the only thing dask will transfer is the result metadata as opposed to the data itself? say for example I do

lots_of_dataframes_in_gcs = task(
    process_data,    
    result=GCSResult(
        PandasSerializer(file_type="parquet")
        )
    )
).map(input_files)
task(notify_data_is_ready).set_dependencies(upstream_tasks=[lots_of_dataframes])

I don't actually want the data passed to the downstream task, just to enforce the ordering. but based on my exploding daskworker memory it seems that the downstream task is actually collecting all of the outputs themselves(!), even though each transferred object shows up in the dask dashboard as a 48-byte

Cached

state object (so the memory is 99.999% untracked by dask) I can imagine a workaround that involves just skipping Results altogether and returning True or a file path explicitly, but it seems like this might be a bug and the

state.result

should be getting purged before we ship it around? @Chris White @Zanie or anyone else have thoughts on whether this is expected?

Interesting -- the data is transferred because it is attached to the upstream state object

because it is attached to the upstream state object

by necessity or just bc it happens to still be there...? seems like the Result would be serving that purpose if one is set (if not then obviously it needs to hold on to to the data still).

in memory movement of data without the back and forth trip is typically more of a feature than a nuisance so attaching the data directly to states is more necessity than not; do any downstream tasks from this one use the data directly?

no, it's quite impossible to load all of the data on one worker (many TB) so all the job is doing downstream is loading to BigQuery and notifying on success. I actually had started hacking together my own result for this use case but never quite got it working, had hoped to just stick with the built-in stuff and use

set_dependencies

but that doesn't actually work

+class GCSReference(GCSResult):
+    """Prefect Result class that returns a path to the data instead of the data itself.
+
+    Useful for very large results that can't be loaded locally, e.g. mapped task outputs that will
+    be loaded into BigQuery without leaving GCS.
+    """
+
+    def read(self, location: str) -> Result:
+        new = self.copy()
+        new.location = location
+        new.value = location
+        return new

any reason you can't instantiate the

GCSResult

class within the task and call its write method?

I wouldn't get checkpointing though if I don't use a real Result object though, right?

Yea exactly exactly