Build a custom Dagster IO manager to persist asset outputs to a specific storage backend (e.g., Parquet on S3)

Steps

1. Subclass IOManager and implement handle_output(context, obj) to write the asset's output and load_input(context) to read it back
2. Use context.asset_key, context.partition_key, and context.metadata to construct a deterministic storage path such as s3://bucket/asset_name/partition_key/data.parquet
3. Register the IO manager as a resource in the Definitions object under 'io_manager' or a named key, then reference it per-asset with io_manager_key='my_io_manager'
4. Return the loaded object from load_input() as the expected Python type; use context.dagster_type to validate type compatibility at runtime
5. Add retry logic inside handle_output for transient storage errors; Dagster will not automatically retry IO manager calls
6. Test the IO manager with build_input_context and build_output_context helpers to verify read/write behavior without running a full Dagster pipeline