step-3-io-management.md

Step 3: IO management

Define the CSV IO manager

class CSVIOManager(ConfigurableIOManager):
    def handle_output(self, context: OutputContext, obj: pd.DataFrame) -> None:
        ...
    def load_input(self, context: InputContext) -> pd.DataFrame:
        ...

  def handle_output(self, context: OutputContext, obj: pd.DataFrame) -> None:
      path = get_path(context)
      obj.to_csv(path)

  def load_input(self, context: InputContext) -> pd.DataFrame:
      path = get_path(context)
      return pd.read_csv(path)

Register the IO manager

from ames_housing.io_managers.csv_io_manager import CSVIOManager

definitions = Definitions(
    ...
    resources={
        "csv_io_manager": CSVIOManager(),
    },
)

Set the IO manager key on the assets

@asset(io_manager_key="csv_io_manager")

@multi_asset(
    outs={
        "train_data": AssetOut(io_manager_key="csv_io_manager"),
        "test_data": AssetOut(io_manager_key="csv_io_manager"),
    }
)