def METHOD_NAME(task, inputs, output, local=False, cwd=None, force=True, writer_opts=None,
        copy_single=False):
    """
    This method is intended to be used by tasks that are supposed to merge parquet files, e.g. when
    inheriting from :py:class:`law.contrib.tasks.MergeCascade`. *inputs* should be a sequence of
    targets that represent the files to merge into *output*.
    When *local* is *False* and files need to be copied from remote first, *cwd* can be a set as the
    dowload directory. When empty, a temporary directory is used. The *task* itself is used to print
    and publish messages via its :py:meth:`law.Task.publish_message` and
    :py:meth:`law.Task.publish_step` methods. When *force* is *True*, any existing output file is
    overwritten.
    *writer_opts* and *copy_single* are forwarded to :py:func:`merge_parquet_files` which is used
    internally for the actual merging.
    """
    abspath = lambda path: os.path.abspath(os.path.expandvars(os.path.expanduser(path)))
    # ensure inputs are targets
    inputs = [
        LocalFileTarget(abspath(inp)) if isinstance(inp, six.string_types) else inp
        for inp in inputs
    ]
    # ensure output is a target
    if isinstance(output, six.string_types):
        output = LocalFileTarget(abspath(output))
    def merge(inputs, output):
        with task.publish_step("merging {} parquet files ...".format(len(inputs)), runtime=True):
            # clear the output if necessary
            if output.exists() and force:
                output.remove()
            # merge
            merge_parquet_files(
                [inp.path for inp in inputs],
                output.path,
                writer_opts=writer_opts,
                copy_single=copy_single,
            )
        stat = output.exists(stat=True)
        if not stat:
            raise Exception("output '{}' not creating during merging".format(output.path))
        # print the size
        output_size = human_bytes(stat.st_size, fmt=True)
        task.publish_message("merged file size: {}".format(output_size))
    if local:
        # everything is local, just merge
        merge(inputs, output)
    else:
        # when not local, we need to fetch files first into the cwd
        if not cwd:
            cwd = LocalDirectoryTarget(is_tmp=True)
        elif isinstance(cwd, str):
            cwd = LocalDirectoryTarget(abspath(cwd))
        cwd.touch()
        # fetch
        with task.publish_step("fetching inputs ...", runtime=True):
            def fetch(inp):
                local_inp = cwd.child(inp.unique_basename, type="f")
                inp.copy_to_local(local_inp, cache=False)
                return local_inp
            def callback(i):
                task.publish_message("fetch file {} / {}".format(i + 1, len(inputs)))
            local_inputs = map_verbose(fetch, inputs, every=5, callback=callback)
        # merge into a localized output
        with output.localize("w", cache=False) as local_output:
            merge(local_inputs, local_output)