This looks like a serialization/deserialization issue with persisted cached results, not a dbt-specific failure. Based on Prefect 3.x behavior, the most likely cause is: * the cached result is being written to S3 with a pickle/base64 serializer path * on the second run, Prefect reads that cached object back and fails decoding it * the error suggests the stored payload is no longer valid base64 in the format the loader expects So the cache write succeeds, but the cache read fails. What I’d try first 1. Switch the result serializer away from pickle/base64 2. Clear the existing corrupted cache objects in S3 3. Re-run A good first option is

JSONSerializer

or

CompressedJSONSerializer

, if your cached values are JSON-serializable. Example at the flow level:

from prefect import flow
from prefect.serializers import JSONSerializer

@flow(
    persist_result=True,
    result_serializer=JSONSerializer(),
)
def my_flow():
    ...

Or:

from prefect import flow
from prefect.serializers import CompressedJSONSerializer

@flow(
    persist_result=True,
    result_serializer=CompressedJSONSerializer(),
)
def my_flow():
    ...

If the orchestrator/task is what’s being cached, apply the serializer where that result persistence is configured. Important: after changing serializers, delete the old cached objects from that S3 cache location. Otherwise Prefect may keep trying to read the already-bad cached payloads. Why this happens Prefect’s result persistence path can involve base64-encoded bytes for pickle-based serializers. The failure you’re seeing on the second run is consistent with a previously persisted cache entry being read back as an invalid base64 payload. So the pattern is: * first run: writes cache to S3 * second run: attempts cache hit * deserialize fails with

Invalid base64-encoded string...

Recommended workaround options Best option * Use

JSONSerializer()

or

CompressedJSONSerializer()

for cached/persisted results Other options * Temporarily disable caching to confirm this is the read path * Point cache/result storage somewhere local and compare behavior * Clear the S3 cache prefix entirely and retry from a clean state Quick debugging steps 1. Confirm whether the failure only happens when there is already cached data 2. Delete the S3 cache prefix 3. Re-run once 4. Re-run again 5. If it only fails on the second run, that strongly confirms a bad persisted cache record Docs * Prefect results * Prefect task caching * prefect-dbt integration docs If you want, send me: * the snippet where you configure

PrefectDbtOrchestrator

* where you set the cache policy / result storage * whether you’re using

persist_result=True

* your serializer config, if any and I can help pinpoint exactly where to change it.

Thanks — that helps. Given this setup, I’d expect the issue to be in the orchestrator cache persistence layer itself, specifically when reading cached node results back from S3 on a later run. Your config:

orchestrator = PrefectDbtOrchestrator(
    settings=settings,
    execution_mode=ExecutionMode.PER_NODE,
    concurrency=1,
    threads=2,
    retries=3,
    retry_delay_seconds=60,
    cache=CacheConfig(
        result_storage="s3-bucket/...",
        key_storage="s3-bucket/...",
    ),
)

A few important notes: *

PrefectDbtOrchestrator

cache is not the same as standard

@task(cache_policy=...)

* the failure pattern you described strongly suggests cached node results are being serialized to S3 in a way that cannot be decoded on reuse * because it fails on the second run, the write path is likely fine, and the read/deserialization path is the problem Most likely fix Try changing one or both cache locations off S3 first to confirm the issue:

orchestrator = PrefectDbtOrchestrator(
    settings=settings,
    execution_mode=ExecutionMode.PER_NODE,
    concurrency=1,
    threads=2,
    retries=3,
    retry_delay_seconds=60,
    cache=CacheConfig(
        result_storage="/tmp/prefect-dbt-results",
        key_storage="/tmp/prefect-dbt-keys",
    ),
)

If local storage works across repeated runs, then this is very likely an S3-backed cache serialization issue rather than a dbt issue. Very important: your

result_storage

/

key_storage

values should be valid URIs/paths for what the orchestrator expects. If you literally have:

"s3-bucket/..."

instead of something like:

"<s3://my-bucket/some/prefix>"

that could also matter. I’d double-check that the exact values are fully qualified. What I’d test next Test 1: clear the cache and retry Delete the existing objects under both prefixes, then run twice again. Test 2: use local paths If local works and S3 fails, that isolates the issue to S3 cache persistence. Test 3: split storages Try local

key_storage

with S3

result_storage

, then local

result_storage

with S3 `key_storage`:

cache=CacheConfig(
    result_storage="<s3://my-bucket/dbt-cache/results>",
    key_storage="/tmp/prefect-dbt-keys",
)

and then:

cache=CacheConfig(
    result_storage="/tmp/prefect-dbt-results",
    key_storage="<s3://my-bucket/dbt-cache/keys>",
)

That will tell us whether the corrupt payload is in: * cached result objects, or * cache key metadata My suspicion I suspect

result_storage

is the culprit, because the base64 error sounds like a persisted result payload being decoded incorrectly. Short-term workaround If you need to move forward quickly, I’d use: * local cache storage, or * disable orchestrator cache temporarily until you can isolate whether this is specific to S3-backed orchestrator caching. Docs * prefect-dbt integration * Prefect results If you want, I can help you build a minimal repro matrix for: * local/local * s3/local * local/s3 * s3/s3 so you can determine exactly which storage path is breaking.