Has anyone run into any problems reading a result written with a serializer other than the default pickle serializer? I set the result for a task to use GCSResult with the JSONSerializer. When I try to read the result from task runs obtained from

Client.get_flow_run_info(..)

, the read function is using the default pickle serializer. I used prefect 0.15.1 in both the flow environment and to run the read code. Am I missing something? Thank you in advance!

result = GCSResult(
    bucket="prefect-cloud-results",
    location="{flow_name}/{flow_run_id}/provenance.json",
    serializer=JSONSerializer()
)
@task(result=result)
def set_provenance_data(flow_run_id: str, prefect_cloud_client: prefect.client.client.Client = None):
...

Hey @Ayla Khan , I have seen this happen specifically when the serialization and deserialization happened with different Python versions. Both Python versions and Prefect versions (a bit less on this one) need to match on both ends. Are you using a container that has a different Python version (3.6, 3.7, 3.8)?

Yes, they're different. Thank you - I'll try a container with a different Python version and see if that fixes it!

Could you give me your

cloudpickle

version? I’ll make an environment and test this

Thank you!

root@893f541d47cb:/# pip list | grep cloudpickle
cloudpickle                 1.6.0

Will try to replicate sometime today. Feel free to ping me if I don’t get back you by tom. I just have a bunch of stuff today.

ok, sounds good

I know what you are saying now. I can replicate that error when I serialize with JSON and read with PickleSerializer. Did you attach the Result to the class or use it inside the task? Did you attach it to the Flow?

I want to return the result from the task and use it (attach I guess?) inside a Flow

I mean did you do

@task(result = xxx)

or

Flow(result = xxx)

or

@task
def abc():
    result.write(xxx)

Wondering how the result was defined.

Ah, ok! I'm doing

@task(result = xxx)

So I talked to the team. The serializer is not included in the result by design. It is loaded in when the Task runs, so it won’t work for the script that you posted originally. What are you trying to do?

I have some data about flow runs that for ergonomic reasons, is better to serialize as JSON instead of pickling, and save that JSON file to a GCS bucket. It's also convenient to have this data available to query through

Client.get_flow_run_info

and the task run data that gets returned, which is what I was testing in the script snippet I posted. I have a workaround by calling raw google storage code directly, but it would be nice to be able to get that data by reading from the result. Is there any way to specify what serializer a result should use when reading a value from a previously run task state's result? Would I set a Result on a task or flow that's instantiated with the serializer the result should use and then call the result read function?

Yeah so I was thinking maybe you should just instantiate the Result manually like

res = GCSResult(bucket="xxx", serializer=JSONSerializer())

, and then you can do

res.read(location)

Ok, I'll give that a try later today. Thank you!

Thanks for your understanding!