diff --git a/dvc/remote/hdfs.py b/dvc/remote/hdfs.py index 337b777b1..7b2084048 100644 --- a/dvc/remote/hdfs.py +++ b/dvc/remote/hdfs.py @@ -9,7 +9,7 @@ from subprocess import Popen, PIPE import dvc.logger as logger from dvc.config import Config from dvc.remote.base import RemoteBase, RemoteCmdError -from dvc.utils import fix_env +from dvc.utils import fix_env, tmp_fname class RemoteHDFS(RemoteBase): @@ -118,13 +118,22 @@ class RemoteHDFS(RemoteBase): if from_info["scheme"] != "local": raise NotImplementedError - cmd = "mkdir -p {}".format(posixpath.dirname(to_info["path"])) - self.hadoop_fs(cmd, user=to_info["user"]) + self.hadoop_fs( + "mkdir -p {}".format(posixpath.dirname(to_info["path"])), + user=to_info["user"], + ) - cmd = "copyFromLocal {} {}".format( - from_info["path"], to_info["path"] + tmp_file = tmp_fname(to_info["user"]) + + self.hadoop_fs( + "copyFromLocal {} {}".format(from_info["path"], tmp_file), + user=to_info["user"], + ) + + self.hadoop_fs( + "mv {} {}".format(tmp_file, to_info["path"]), + user=to_info["user"], ) - self.hadoop_fs(cmd, user=to_info["user"]) def download( self, @@ -151,10 +160,14 @@ class RemoteHDFS(RemoteBase): if not os.path.exists(dname): os.makedirs(dname) - cmd = "copyToLocal {} {}".format( - from_info["path"], to_info["path"] + tmp_file = tmp_fname(to_info["user"]) + + self.hadoop_fs( + "copyToLocal {} {}".format(from_info["path"], tmp_file), + user=from_info["user"], ) - self.hadoop_fs(cmd, user=from_info["user"]) + + os.rename(tmp_file, to_info["path"]) def list_cache_paths(self): try: diff --git a/dvc/remote/local.py b/dvc/remote/local.py index 7a81b9704..cf5f3b360 100644 --- a/dvc/remote/local.py +++ b/dvc/remote/local.py @@ -486,9 +486,11 @@ class RemoteLOCAL(RemoteBase): name = os.path.basename(from_info["path"]) makedirs(os.path.dirname(to_info["path"]), exist_ok=True) + tmp_file = tmp_fname(to_info["path"]) try: - copyfile(from_info["path"], to_info["path"], name=name) + copyfile(from_info["path"], tmp_file, name=name) + os.rename(tmp_file, to_info["path"]) except Exception: logger.error( "failed to upload '{}' to '{}'".format( @@ -531,6 +533,8 @@ class RemoteLOCAL(RemoteBase): no_progress_bar=no_progress_bar, name=name, ) + + os.rename(tmp_file, to_info["path"]) except Exception: logger.error( "failed to download '{}' to '{}'".format( @@ -540,8 +544,6 @@ class RemoteLOCAL(RemoteBase): continue - os.rename(tmp_file, to_info["path"]) - def _group(self, checksum_infos, show_checksums=False): by_md5 = {}