Add native hdfs support for gz (#128)

* use native support for gz

* add readline for hdfs

smart_open/smart_open_lib.py

@@ -492,14 +492,24 @@ def __init__(self, parsed_uri):
         if parsed_uri.scheme not in ("hdfs"):
             raise TypeError("can only process HDFS files")
         self.parsed_uri = parsed_uri
+        self._readline_iter = None
 
     def __iter__(self):
-        hdfs = subprocess.Popen(["hdfs", "dfs", "-cat", self.parsed_uri.uri_path], stdout=subprocess.PIPE)
+        hdfs = subprocess.Popen(["hdfs", "dfs", '-text', self.parsed_uri.uri_path], stdout=subprocess.PIPE)
         return hdfs.stdout
 
     def read(self, size=None):
         raise NotImplementedError("read() not implemented yet")
 
+    def readline(self):
+        if self._readline_iter is None:
+            self._readline_iter = self.__iter__()
+        try:
+            return next(self._readline_iter)
+        except StopIteration:
+            # When readline runs out of data, it just returns an empty string
+            return ''
+
     def seek(self, offset, whence=None):
         raise NotImplementedError("seek() not implemented yet")
 

smart_open/tests/test_smart_open.py

@@ -216,12 +216,12 @@ def test_hdfs(self, mock_subprocess):
         smart_open_object = smart_open.HdfsOpenRead(smart_open.ParseUri("hdfs:///tmp/test.txt"))
         smart_open_object.__iter__()
         # called with the correct params?
-        mock_subprocess.Popen.assert_called_with(["hdfs", "dfs", "-cat", "/tmp/test.txt"], stdout=mock_subprocess.PIPE)
+        mock_subprocess.Popen.assert_called_with(["hdfs", "dfs", "-text", "/tmp/test.txt"], stdout=mock_subprocess.PIPE)
 
         # second possibility of schema
         smart_open_object = smart_open.HdfsOpenRead(smart_open.ParseUri("hdfs://tmp/test.txt"))
         smart_open_object.__iter__()
-        mock_subprocess.Popen.assert_called_with(["hdfs", "dfs", "-cat", "/tmp/test.txt"], stdout=mock_subprocess.PIPE)
+        mock_subprocess.Popen.assert_called_with(["hdfs", "dfs", "-text", "/tmp/test.txt"], stdout=mock_subprocess.PIPE)
 
     @responses.activate
     def test_webhdfs(self):