diff --git a/dvc/tree/hdfs.py b/dvc/tree/hdfs.py index 2c2db68c3..eb6b5f449 100644 --- a/dvc/tree/hdfs.py +++ b/dvc/tree/hdfs.py @@ -2,6 +2,7 @@ import io import logging import os import re +import shutil import subprocess from collections import deque from contextlib import closing, contextmanager @@ -162,7 +163,7 @@ class HDFSTree(BaseTree): with closing( hdfs.open_output_stream(tmp_info.path) ) as tmp_fobj: - tmp_fobj.write(from_fobj.read()) + shutil.copyfileobj(from_fobj, tmp_fobj) hdfs.move(tmp_info.path, to_info.path) except Exception: self.remove(tmp_info) diff --git a/dvc/tree/webhdfs.py b/dvc/tree/webhdfs.py index a65568f09..2544dd7a7 100644 --- a/dvc/tree/webhdfs.py +++ b/dvc/tree/webhdfs.py @@ -1,5 +1,6 @@ import logging import os +import shutil import threading from contextlib import contextmanager from urllib.parse import urlparse @@ -134,8 +135,8 @@ class WebHDFSTree(BaseTree): def copy(self, from_info, to_info, **_kwargs): with self.hdfs_client.read(from_info.path) as reader: - content = reader.read() - self.hdfs_client.write(to_info.path, data=content) + with self.hdfs_client.write(to_info.path) as writer: + shutil.copyfileobj(reader, writer) def move(self, from_info, to_info, mode=None): self.hdfs_client.makedirs(to_info.parent.path)