remote: make sure we are using a temp file to make the upload/download atomic
Same as with https://github.com/iterative/dvc/issues/1570 , but now we've caught it with our local remote. Need to go through all of our remotes that support upload/download and make sure that they are using temporary files to make the operation atomic. S3, GS and Azure do that for us when uploading, so no need to worry about them.

https://discordapp.com/channels/485586884165107732/485596304961962003/551078399083544578
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"],
+            )
+
+            tmp_file = tmp_fname(to_info["user"])
+
+            self.hadoop_fs(
+                "copyFromLocal {} {}".format(from_info["path"], tmp_file),
+                user=to_info["user"],
+            )
 
-            cmd = "copyFromLocal {} {}".format(
-                from_info["path"], to_info["path"]
+            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 = {}
 
['dvc/remote/local.py', 'dvc/remote/hdfs.py']