def METHOD_NAME(
    script: Union[bytes, str, BinaryIO, TextIO],
    data: Dict[str, TileableType] = None,
    n_workers: int = 1,
    command_argv: List[str] = None,
    session: SessionType = None,
    retry_when_fail: bool = False,
    run_kwargs: Dict[str, Any] = None,
):
    """
    Run script in Mars cluster.
    Parameters
    ----------
    script: str or file-like object
        Script to run.
    data: dict
        Variable name to data.
    n_workers: int
        number of workers to run the script
    command_argv: list
        extra command args for script
    session: Mars session
        if not provided, will use default one
    retry_when_fail: bool, default False
       If True, retry when function failed.
    run_kwargs: dict
        extra kwargs for session.run
    Returns
    -------
    Object
        Mars Object.
    """
    if hasattr(script, "read"):
        code = script.read()
    else:
        with open(os.path.abspath(script), "rb") as f:
            code = f.read()
    inputs = _extract_inputs(data)
    op = RunScript(
        data=data,
        code=to_binary(code),
        world_size=n_workers,
        retry_when_fail=retry_when_fail,
        command_args=command_argv,
    )
    return op(inputs).execute(session=session, **(run_kwargs or {}))