Great question! Since you’re on Prefect 2.x (Cloud) and running on GCP, the most common way to use

JSONSerializer

is to persist task (and/or flow) results to a GCS bucket via a GcsBucket block, with JSON as the format. Here’s a minimal example:

# pip install prefect-gcp
from prefect import flow, task
from prefect.serializers import JSONSerializer
from prefect_gcp.cloud_storage import GcsBucket

# In Prefect Cloud, create a GcsBucket block (e.g., named "my-gcs")
# that points to your bucket and has credentials configured.

@task(
    persist_result=True,  # actually write the result to storage
    result_storage=GcsBucket.load("my-gcs"),
    result_serializer=JSONSerializer(),  # store as JSON
)
def compute():
    return {"a": 1, "b": ["x", "y"], "note": "stored as JSON in GCS"}

@flow
def main():
    r = compute()
    return r  # Optional: you can also persist the flow return value (see below)

if __name__ == "__main__":
    main()

Notes and options: - You can set JSON serialization at the flow level too:

@flow(
      result_storage=GcsBucket.load("my-gcs"),
      result_serializer=JSONSerializer(),
      persist_result=True,
  )
  def main():
      ...

The flow’s own return value will be persisted; tasks still need

persist_result=True

if you want their outputs persisted. - If you want a smaller on-disk footprint, use

CompressedJSONSerializer()

instead. -

JSONSerializer

uses Prefect’s object encoder/decoder, which supports common types like datetimes, UUIDs, and Pydantic models. If your objects are still not JSON-serializable, consider

PickleSerializer

as a fallback. - You can customize JSON output:

JSONSerializer(dumps_kwargs={"indent": 2, "sort_keys": True})

Manual use (outside of task/flow config):

from prefect.serializers import JSONSerializer

serializer = JSONSerializer()
data = {"x": 1}
payload = serializer.dumps(data)  # bytes or str payload
restored = serializer.loads(payload)

Helpful links: - Results (Prefect 2): https://docs-v2.prefect.io/concepts/results/ - GCS block (prefect-gcp): https://prefecthq.github.io/prefect-gcp/cloud_storage/#prefect_gcp.cloud_storage.GcsBucket - prefect-gcp collection: https://prefecthq.github.io/prefect-gcp/ If you share your block name and how you’re running flows (local, agent in GCP, etc.), I can tailor the example further.

You’re hitting this because Prefect’s JSONSerializer doesn’t handle NumPy types by default. You have a few good options: 1) Convert NumPy to built-in Python types before returning - Easiest and most explicit.

import numpy as np
from prefect import task

@task(persist_result=True)
def compute():
    x = np.int64(42)
    arr = np.array([1, 2, 3])
    return {"x": int(x), "arr": arr.tolist()}  # cast scalars, use .tolist() for arrays

2) Provide a custom object_encoder that knows NumPy - Define a module-level function and reference it by its fully-qualified name.

# my_project/encoders.py
import numpy as np
from prefect.serializers import prefect_json_object_encoder

def numpy_json_encoder(obj):
    if isinstance(obj, np.integer):
        return int(obj)
    if isinstance(obj, np.floating):
        return float(obj)
    if isinstance(obj, np.bool_):
        return bool(obj)
    if isinstance(obj, np.ndarray):
        return obj.tolist()
    return prefect_json_object_encoder(obj)  # fallback for other types

Use it with JSONSerializer:

from prefect import task
from prefect.serializers import JSONSerializer
from prefect_gcp.cloud_storage import GcsBucket

@task(
    persist_result=True,
    result_storage=GcsBucket.load("my-gcs"),
    result_serializer=JSONSerializer(
        object_encoder="my_project.encoders.numpy_json_encoder"
    ),
)
def compute():
    ...

Notes: - The encoder must be importable on the worker (i.e., code deployed with your flow). - Don’t pass `dumps_kwargs={"default": ...}`; Prefect 2.x forbids that. Use

object_encoder

instead. 3) Use orjson with NumPy support - orjson can serialize many NumPy types if you enable the option.

# pip install orjson
import orjson
from prefect.serializers import JSONSerializer

serializer = JSONSerializer(
    jsonlib="orjson",
    dumps_kwargs={"option": orjson.OPT_SERIALIZE_NUMPY},
)

Then set

result_serializer=serializer

on your task/flow. Make sure orjson is installed in the runtime environment. 4) If JSON isn’t required, use PickleSerializer - Most flexible, but opaque (not human-readable).

from prefect.serializers import PickleSerializer
@task(persist_result=True, result_serializer=PickleSerializer())
def compute():
    return {"x": np.int64(42)}

Links: - Prefect 2 results/serialization: https://docs-v2.prefect.io/concepts/results/ - JSON serializer source (2.x): https://github.com/PrefectHQ/prefect/blob/2.x/src/prefect/serializers.py - orjson and NumPy: https://github.com/ijl/orjson#numpy If you share what kinds of NumPy objects you’re returning (scalars, arrays, dtypes), I can suggest the most streamlined option for your case.

Short answer: set a custom storage key on the task with

result_storage_key=...

. Flows don’t support custom filenames; tasks do. Example with JSONSerializer and GCS:

from prefect import flow, task
from prefect.serializers import JSONSerializer
from prefect_gcp.cloud_storage import GcsBucket

gcs = GcsBucket.load("my-gcs")

@task(
    persist_result=True,
    result_storage=gcs,
    result_serializer=JSONSerializer(),
    result_storage_key="results/{flow_run.flow_name}/{parameters.job_id}.json",
)
def compute(job_id: int):
    return {"job_id": job_id, "status": "ok"}

@flow
def main(job_id: int):
    return compute(job_id)

if __name__ == "__main__":
    main(123)

What this does: - The task result is stored at gs://<your-bucket>/<optional-bucket_folder>/results/<flow_name>/123.json -

result_storage_key

supports templating with runtime values like

parameters[...]

,

flow_run.flow_name

,

flow_run.name

, etc. - If you don’t specify

result_storage_key

, Prefect uses a random UUID filename. Notes: - Only tasks can set `result_storage_key`; the flow return value’s filename is not configurable. - If you set a

bucket_folder

on the GcsBucket block, it’s prepended to the key. - If you expect collisions, include unique parts (e.g.,

{flow_run.name}

or a timestamp) to avoid overwrites. If you share your desired path pattern, I can help craft the exact template.