Function wrap_reduce_function_for_file_system

pandarallel/core.py:175–201

This wrapper transforms a `reduce` function which takes as input: - A list of pandas Dataframe - An user defined function and which returns a pandas Dataframe, into a `reduct` function which takes as input: - A list of paths where pandas Dataframe are pickled which returns a pan

(
    reduce_function: Callable[[Iterator, Dict[str, Any]], Any]
)

def wrap_reduce_function_for_file_system(
    reduce_function: Callable[[Iterator, Dict[str, Any]], Any]
) -> Callable[[Iterator[Path], Dict[str, Any]], Any]:
    """This wrapper transforms a `reduce` function which takes as input:
    - A list of pandas Dataframe
    - An user defined function
    and which returns a pandas Dataframe, into a `reduct` function which takes as input:
    - A list of paths where  pandas Dataframe are pickled
    which returns a pandas Dataframe.
    """

    def closure(output_file_paths: Iterator[Path], extra: Dict[str, Any]) -> Any:
        def get_dataframe_and_delete_file(file_path: Path) -> Any:
            with file_path.open("rb") as file_descriptor:
                data = pickle.load(file_descriptor)

            file_path.unlink()
            return data

        dfs = (
            get_dataframe_and_delete_file(output_file_path)
            for output_file_path in output_file_paths
        )

        return reduce_function(dfs, extra)

    return closure


def parallelize_with_memory_file_system(