Browse Source

[webhdfs] Add copy method to copy a file

bc Wong 13 years ago
parent
commit
c43cd8e864

+ 27 - 2
desktop/libs/hadoop/src/hadoop/fs/fs_test.py

@@ -15,12 +15,15 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import logging
 import os
+import stat
 import tempfile
 import unittest
-import logging
 
-from hadoop import fs
+from hadoop import fs, pseudo_hdfs4
+from nose.plugins.attrib import attr
+from nose.tools import assert_equal
 
 logger = logging.getLogger(__name__)
 
@@ -91,6 +94,28 @@ class LocalSubFileSystemTest(unittest.TestCase):
     # This shouldn't work!
     self.assertRaises(TypeError, self.fs.open, name="/foo", mode="w")
 
+
+@attr('requires_hadoop')
+def test_hdfs_copy():
+  minicluster = pseudo_hdfs4.shared_cluster()
+  minifs = minicluster.fs
+
+  olduser = minifs.setuser(minifs.superuser)
+  minifs.chmod('/', 0777)
+  minifs.setuser(olduser)
+
+  data = "I will not make flatuent noises in class\n" * 2000
+  minifs.create('/copy_test_src', permission=0646, data=data)
+  minifs.create('/copy_test_dst', data="some initial data")
+
+  minifs.copyfile('/copy_test_src', '/copy_test_dst')
+  actual = minifs.read('/copy_test_dst', 0, len(data) + 100)
+  assert_equal(data, actual)
+
+  sb = minifs.stats('/copy_test_dst')
+  assert_equal(0646, stat.S_IMODE(sb.mode))
+
+
 if __name__ == "__main__":
   logging.basicConfig()
   unittest.main()

+ 46 - 2
desktop/libs/hadoop/src/hadoop/fs/webhdfs.py

@@ -22,6 +22,7 @@ Interfaces for Hadoop filesystem access via HttpFs/WebHDFS
 import errno
 import logging
 import random
+import stat
 import threading
 
 from django.utils.encoding import smart_str
@@ -265,7 +266,11 @@ class WebHdfs(Hdfs):
     self._root.put(path, params)
 
   def chmod(self, path, mode):
-    """chmod(path, mode)"""
+    """
+    chmod(path, mode)
+
+    `mode' should be an octal integer or string.
+    """
     path = Hdfs.normpath(path)
     params = self._getparams()
     params['op'] = 'SETPERMISSION'
@@ -293,7 +298,13 @@ class WebHdfs(Hdfs):
     params['length'] = long(length)
     if bufsize is not None:
       params['bufsize'] = bufsize
-    return self._root.get(path, params)
+    try:
+      return self._root.get(path, params)
+    except WebHdfsException, ex:
+      if "out of the range" in ex.message:
+        return ""
+      raise ex
+      
 
   def open(self, path, mode='r'):
     """
@@ -312,6 +323,7 @@ class WebHdfs(Hdfs):
     create(path, overwrite=False, blocksize=None, replication=None, permission=None)
 
     Creates a file with the specified parameters.
+    `permission' should be an octal integer or string.
     """
     path = Hdfs.normpath(path)
     params = self._getparams()
@@ -338,6 +350,38 @@ class WebHdfs(Hdfs):
     params['op'] = 'APPEND'
     self._invoke_with_redirect('POST', path, params, data)
 
+
+  def copyfile(self, src, dst):
+    sb = self._stats(src)
+    if sb is None:
+      raise IOError(errno.ENOENT, "Copy src '%s' does not exist" % (src,))
+    if sb.isDir:
+      raise IOError(errno.INVAL, "Copy src '%s' is a directory" % (src,))
+    if self.isdir(dst):
+      raise IOError(errno.INVAL, "Copy dst '%s' is a directory" % (dst,))
+
+    CHUNK_SIZE = 65536
+    offset = 0
+    
+    while True:
+      data = self.read(src, offset, CHUNK_SIZE)
+      if offset == 0:
+        self.create(dst,
+                    overwrite=True,
+                    blocksize=sb.blockSize,
+                    replication=sb.replication,
+                    permission=oct(stat.S_IMODE(sb.mode)),
+                    data=data)
+
+      cnt = len(data)
+      if cnt == 0:
+        break
+
+      if offset != 0:
+        self.append(dst, data)
+      offset += cnt
+
+
   @staticmethod
   def urlsplit(url):
     return Hdfs.urlsplit(url)