FM / utils

/usr/local/lib/python3.6/site-packages/datasets/utils/

/usr/local/lib/python3.6/site-packages/datasets/utils

Name	Size	Mode	Actions
resources/	-	0755	rm
__pycache__/	-	0755	rm
beam_utils.py	2029	0644	edit dl rm
deprecation_utils.py	2830	0644	edit dl rm
doc_utils.py	407	0644	edit dl rm
download_manager.py	60	0644	edit dl rm
extract.py	10885	0644	edit dl rm
filelock.py	13813	0644	edit dl rm
file_utils.py	23120	0644	edit dl rm
info_utils.py	3813	0644	edit dl rm
logging.py	6762	0644	edit dl rm
metadata.py	21558	0644	edit dl rm
patching.py	4957	0644	edit dl rm
py_utils.py	38136	0644	edit dl rm
readme.py	12671	0644	edit dl rm
stratify.py	4091	0644	edit dl rm
tf_utils.py	2093	0644	edit dl rm
typing.py	176	0644	edit dl rm
version.py	4249	0644	edit dl rm
_hf_hub_fixes.py	3808	0644	edit dl rm
__init__.py	921	0644	edit dl rm

Edit: /usr/local/lib/python3.6/site-packages/datasets/utils/beam_utils.py (2029B)

import os

from apache_beam.io.filesystems import FileSystems
from apache_beam.pipeline import Pipeline

from .logging import get_logger

CHUNK_SIZE = 2 << 20  # 2mb
logger = get_logger(__name__)

class BeamPipeline(Pipeline):
    """Wrapper over `apache_beam.pipeline.Pipeline` for convenience"""

def is_local(self):
        runner = self._options.get_all_options().get("runner")
        return runner in [None, "DirectRunner", "PortableRunner"]

def upload_local_to_remote(local_file_path, remote_file_path, force_upload=False):
    """Use the Beam Filesystems to upload to a remote directory on gcs/s3/hdfs..."""
    fs = FileSystems
    if fs.exists(remote_file_path):
        if force_upload:
            logger.info(f"Remote path already exist: {remote_file_path}. Overwriting it as force_upload=True.")
        else:
            logger.info(f"Remote path already exist: {remote_file_path}. Skipping it as force_upload=False.")
            return
    with fs.create(remote_file_path) as remote_file:
        with open(local_file_path, "rb") as local_file:
            chunk = local_file.read(CHUNK_SIZE)
            while chunk:
                remote_file.write(chunk)
                chunk = local_file.read(CHUNK_SIZE)

def download_remote_to_local(remote_file_path, local_file_path, force_download=False):
    """Use the Beam Filesystems to download from a remote directory on gcs/s3/hdfs..."""
    fs = FileSystems
    if os.path.exists(local_file_path):
        if force_download:
            logger.info(f"Local path already exist: {remote_file_path}. Overwriting it as force_upload=True.")
        else:
            logger.info(f"Local path already exist: {remote_file_path}. Skipping it as force_upload=False.")
            return
    with fs.open(remote_file_path) as remote_file:
        with open(local_file_path, "wb") as local_file:
            chunk = remote_file.read(CHUNK_SIZE)
            while chunk:
                local_file.write(chunk)
                chunk = remote_file.read(CHUNK_SIZE)

import os from apache_beam.io.filesystems import FileSystems from apache_beam.pipeline import Pipeline from .logging import get_logger CHUNK_SIZE = 2 << 20 # 2mb logger = get_logger(__name__) class BeamPipeline(Pipeline): """Wrapper over `apache_beam.pipeline.Pipeline` for convenience""" def is_local(self): runner = self._options.get_all_options().get("runner") return runner in [None, "DirectRunner", "PortableRunner"] def upload_local_to_remote(local_file_path, remote_file_path, force_upload=False): """Use the Beam Filesystems to upload to a remote directory on gcs/s3/hdfs...""" fs = FileSystems if fs.exists(remote_file_path): if force_upload: logger.info(f"Remote path already exist: {remote_file_path}. Overwriting it as force_upload=True.") else: logger.info(f"Remote path already exist: {remote_file_path}. Skipping it as force_upload=False.") return with fs.create(remote_file_path) as remote_file: with open(local_file_path, "rb") as local_file: chunk = local_file.read(CHUNK_SIZE) while chunk: remote_file.write(chunk) chunk = local_file.read(CHUNK_SIZE) def download_remote_to_local(remote_file_path, local_file_path, force_download=False): """Use the Beam Filesystems to download from a remote directory on gcs/s3/hdfs...""" fs = FileSystems if os.path.exists(local_file_path): if force_download: logger.info(f"Local path already exist: {remote_file_path}. Overwriting it as force_upload=True.") else: logger.info(f"Local path already exist: {remote_file_path}. Skipping it as force_upload=False.") return with fs.open(remote_file_path) as remote_file: with open(local_file_path, "wb") as local_file: chunk = remote_file.read(CHUNK_SIZE) while chunk: local_file.write(chunk) chunk = remote_file.read(CHUNK_SIZE)