def METHOD_NAME(root: str, split: Union[Tuple[str], str], language_pair: Tuple[str] = ("de", "en")):
    """Multi30k dataset
    .. warning::
        using datapipes is still currently subject to a few caveats. if you wish
        to use this dataset with shuffling, multi-processing, or distributed
        learning, please see :ref:`this note <datapipes_warnings>` for further
        instructions.
    For additional details refer to https://www.statmt.org/wmt16/multimodal-task.html#task1
    Number of lines per split:
        - train: 29000
        - valid: 1014
        - test: 1000
    Args:
        root: Directory where the datasets are saved. Default: os.path.expanduser('~/.torchtext/cache')
        split: split or splits to be returned. Can be a string or tuple of strings. Default: ('train', 'valid', 'test')
        language_pair: tuple or list containing src and tgt language. Available options are ('de','en') and ('en', 'de')
    :return: DataPipe that yields tuple of source and target sentences
    :rtype: (str, str)
    """
    assert len(language_pair) == 2, "language_pair must contain only 2 elements: src and tgt language respectively"
    assert tuple(sorted(language_pair)) == (
        "de",
        "en",
    ), "language_pair must be either ('de','en') or ('en', 'de')"
    if not is_module_available("torchdata"):
        raise ModuleNotFoundError(
            "Package `torchdata` not found. Please install following instructions at https://github.com/pytorch/data"
        )
    url_dp = IterableWrapper([URL[split]])
    cache_compressed_dp = url_dp.on_disk_cache(
        filepath_fn=partial(_filepath_fn, root, split),
        hash_dict={_filepath_fn(root, split): MD5[split]},
        hash_type="sha256",
    )
    cache_compressed_dp = HttpReader(cache_compressed_dp).end_caching(mode="wb", same_filepath_fn=True)
    cache_compressed_dp_1, cache_compressed_dp_2 = cache_compressed_dp.fork(num_instances=2)
    src_cache_decompressed_dp = cache_compressed_dp_1.on_disk_cache(
        filepath_fn=partial(_decompressed_filepath_fn, root, split, language_pair, 0)
    )
    src_cache_decompressed_dp = (
        FileOpener(src_cache_decompressed_dp, mode="b")
        .load_from_tar()
        .filter(partial(_filter_fn, split, language_pair, 0))
    )
    src_cache_decompressed_dp = src_cache_decompressed_dp.end_caching(mode="wb", same_filepath_fn=True)
    tgt_cache_decompressed_dp = cache_compressed_dp_2.on_disk_cache(
        filepath_fn=partial(_decompressed_filepath_fn, root, split, language_pair, 1)
    )
    tgt_cache_decompressed_dp = (
        FileOpener(tgt_cache_decompressed_dp, mode="b")
        .load_from_tar()
        .filter(partial(_filter_fn, split, language_pair, 1))
    )
    tgt_cache_decompressed_dp = tgt_cache_decompressed_dp.end_caching(mode="wb", same_filepath_fn=True)
    src_data_dp = FileOpener(src_cache_decompressed_dp, encoding="utf-8").readlines(
        return_path=False, strip_newline=True
    )
    tgt_data_dp = FileOpener(tgt_cache_decompressed_dp, encoding="utf-8").readlines(
        return_path=False, strip_newline=True
    )
    return src_data_dp.zip(tgt_data_dp).shuffle().set_shuffle(False).sharding_filter()