trainer

data module

model

train

num_workers=0

num_workers=0

num_workers > 1

num_workers > 1

num_workers > 1

num_worker > 0

num_workers>1

from torch import multiprocessing
def set_multiprocessing_method():
    if multiprocessing.get_start_method() != "spawn":
        multiprocessing.set_start_method(
            "spawn", force=True
        )  # Only SPAWN works with Cuda

TypeError: h5py objects cannot be pickled

__getitem__

__init__

__getitem__

def _load(self) -> tuple[DataSetTorch, Dict]:
   tvcnt = TemporaryVirtualContainer.from_container_paths(
      container_paths=[Path(self._container_paths)],
      dataset_paths=dict(
          data = self._load_args["data_path"],
          target = self._load_args["target_path"],
          mask = self._load_args["mask_path"]
      ),
   )
   dataset = CellPatchDataSetTorch(
       path=tvcnt.path,
       data_path="data",
       target_path="target",
       mask_path="mask",
       data_transform=transforms.Compose([
            cft.DaskToTensor(), 
            cft.FillNans(), 
            cft.MaskImages(), 
            cft.EnsureChannels(num_channels=3)]),
       precision=torch.float32,
   )

vLLM

ParralelFormers

PartitionedDataset

we might need to find a way to pass the initialisation of the parent process for the model training parallelisation to the actual kedro node execution’s main thread.

n=3

def split_df(df:pd.DataFrame)->dict[str,pd.DataFrame]:
    """
    Splits a dataframe into n equal parts
    """
    return {f'df_{i}':df.iloc[i::n] for i in range(n)}

def create_splitted_pipeline(n:int)->Pipeline:
    return pipeline([node(
    func=split_df,
    inputs={'df':'df'},
    outputs={f'df_{i}':f'df_{i}' for i in range(n)},
    name='split_df',
    )])+pipeline([node(
    func=summary,
    inputs={'df':f'df_{i}'},
    outputs={'summary':f'summary_{i}'},
    name=f'summary_{i}') for i in range(n)
    ])