Python bindingds for HDFS filesystem

cpp/src/arrow/filesystem/hdfs.cc

@@ -339,6 +339,12 @@ Result<HdfsOptions> HdfsOptions::FromUri(const Uri& uri) {
   return options;
 }
 
+Result<HdfsOptions> HdfsOptions::FromUri(const std::string& uri_string) {
+  Uri uri;
+  RETURN_NOT_OK(uri.Parse(uri_string));
+  return FromUri(uri);
+}
+
 HadoopFileSystem::HadoopFileSystem(const HdfsOptions& options)
     : impl_(new Impl{options}) {}
 

cpp/src/arrow/filesystem/hdfs.h

@@ -50,6 +50,7 @@ struct ARROW_EXPORT HdfsOptions {
   void ConfigureHdfsBlockSize(int64_t default_block_size);
 
   static Result<HdfsOptions> FromUri(const ::arrow::internal::Uri& uri);
+  static Result<HdfsOptions> FromUri(const std::string& uri);
 };
 
 /// HDFS-backed FileSystem implementation.

python/CMakeLists.txt

@@ -377,6 +377,10 @@ if(PYARROW_BUILD_S3)
   set(CYTHON_EXTENSIONS ${CYTHON_EXTENSIONS} _s3fs)
 endif()
 
+if(PYARROW_BUILD_HDFS)
+  set(CYTHON_EXTENSIONS ${CYTHON_EXTENSIONS} _hdfs)
+endif()
+
 if(PYARROW_BUILD_CUDA)
   # Arrow CUDA
   find_package(ArrowCUDA REQUIRED)

python/pyarrow/_fs.pxd

@@ -22,6 +22,7 @@ import six
 from pyarrow.compat import frombytes, tobytes
 from pyarrow.includes.common cimport *
 from pyarrow.includes.libarrow cimport PyDateTime_from_TimePoint
+from pyarrow.includes.libarrow_fs cimport *
 from pyarrow.lib import _detect_compression
 from pyarrow.lib cimport *
 

python/pyarrow/_hdfs.pyx

@@ -0,0 +1,169 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# cython: language_level = 3
+
+import six
+
+from pyarrow.lib cimport check_status
+from pyarrow.compat import frombytes, tobytes
+from pyarrow.includes.common cimport *
+from pyarrow.includes.libarrow cimport *
+from pyarrow.includes.libarrow_fs cimport *
+from pyarrow._fs cimport FileSystem
+
+
+cdef class HdfsOptions:
+    """Options for HadoopFileSystem.
+
+    Parameters
+    ----------
+
+    """
+    cdef:
+        CHdfsOptions options
+
+    # Avoid mistakingly creating attributes
+    __slots__ = ()
+
+    def __init__(self, endpoint=None, driver=None, replication=None,
+                 user=None, buffer_size=None, default_block_size=None):
+        if endpoint is not None:
+            self.endpoint = endpoint
+        if driver is not None:
+            self.driver = driver
+        if replication is not None:
+            self.replication = replication
+        if user is not None:
+            self.user = user
+        if buffer_size is not None:
+            self.buffer_size = buffer_size
+        if default_block_size is not None:
+            self.default_block_size = default_block_size
+
+    @staticmethod
+    cdef wrap(CHdfsOptions wrapped):
+        cdef HdfsOptions self = HdfsOptions.__new__(HdfsOptions)
+        self.options = wrapped
+        return self
+
+    cdef inline CHdfsOptions unwrap(self) nogil:
+        return self.options
+
+    @staticmethod
+    def from_uri(uri):
+        cdef CResult[CHdfsOptions] result
+        result = CHdfsOptions.FromUriString(tobytes(uri))
+        return HdfsOptions.wrap(GetResultValue(result))
+
+    @property
+    def endpoint(self):
+        return (
+            frombytes(self.options.connection_config.host),
+            self.options.connection_config.port
+        )
+
+    @endpoint.setter
+    def endpoint(self, value):
+        if not isinstance(value, tuple) or len(value) != 2:
+            raise TypeError('Endpoint must be a tuple of host port pair')
+        self.options.connection_config.host = tobytes(value[0])
+        self.options.connection_config.port = tobytes(value[1])
+
+    @property
+    def driver(self):
+        if self.options.connection_config.driver == HdfsDriver_LIBHDFS3:
+            return 'hdfs3'
+        else:
+            return 'hdfs'
+
+    @driver.setter
+    def driver(self, value):
+        if value == 'hdfs3':
+            self.options.connection_config.driver = HdfsDriver_LIBHDFS3
+        elif value == 'hdfs':
+            self.options.connection_config.driver = HdfsDriver_LIBHDFS
+        else:
+            raise ValueError('Choose either hdfs of hdfs3')
+
+    @property
+    def user(self):
+        return frombytes(self.options.connection_config.user)
+
+    @user.setter
+    def user(self, value):
+        self.options.connection_config.user = tobytes(value)
+
+    @property
+    def replication(self):
+        return self.options.replication
+
+    @replication.setter
+    def replication(self, int value):
+        self.options.replication = value
+
+    @property
+    def buffer_size(self):
+        return self.options.buffer_size
+
+    @buffer_size.setter
+    def buffer_size(self, int value):
+        self.options.buffer_size = value
+
+    @property
+    def default_block_size(self):
+        return self.options.default_block_size
+
+    @default_block_size.setter
+    def default_block_size(self, int value):
+        self.options.default_block_size = value
+
+
+cdef class HadoopFileSystem(FileSystem):
+    """HDFS backed FileSystem implementation
+
+    Parameters
+    ----------
+    options: HdfsOptions, default None
+        Options for connecting to hdfs. If None is passed then attempts to
+        initialize the connection from AWS environment variables.
+    """
+
+    cdef:
+        CHadoopFileSystem* hdfs
+
+    def __init__(self, options_or_uri=None):
+        cdef:
+            HdfsOptions options
+            shared_ptr[CHadoopFileSystem] wrapped
+
+        if isinstance(options_or_uri, six.string_types):
+            options = HdfsOptions.from_uri(options_or_uri)
+        elif isinstance(options_or_uri, HdfsOptions):
+            options = options_or_uri
+        elif options_or_uri is None:
+            options = HdfsOptions()
+        else:
+            raise TypeError('Must pass either a string or HdfsOptions')
+
+        with nogil:
+            wrapped = GetResultValue(CHadoopFileSystem.Make(options.unwrap()))
+        self.init(<shared_ptr[CFileSystem]> wrapped)
+
+    cdef init(self, const shared_ptr[CFileSystem]& wrapped):
+        FileSystem.init(self, wrapped)
+        self.hdfs = <CHadoopFileSystem*> wrapped.get()

python/pyarrow/_s3fs.pyx

@@ -23,7 +23,7 @@ from pyarrow.lib cimport check_status
 from pyarrow.compat import frombytes, tobytes
 from pyarrow.includes.common cimport *
 from pyarrow.includes.libarrow cimport *
-from pyarrow.includes.libarrow_s3fs cimport *
+from pyarrow.includes.libarrow_fs cimport *
 from pyarrow._fs cimport FileSystem
 
 
@@ -109,6 +109,9 @@ cdef class S3Options:
         if background_writes is not None:
             self.background_writes = background_writes
 
+    cdef inline CS3Options unwrap(self) nogil:
+        return self.options
+
     @property
     def region(self):
         """AWS region to connect to."""
@@ -166,7 +169,7 @@ cdef class S3FileSystem(FileSystem):
         cdef shared_ptr[CS3FileSystem] wrapped
         options = options or S3Options()
         with nogil:
-            wrapped = GetResultValue(CS3FileSystem.Make(options.options))
+            wrapped = GetResultValue(CS3FileSystem.Make(options.unwrap()))
         self.init(<shared_ptr[CFileSystem]> wrapped)
 
     cdef init(self, const shared_ptr[CFileSystem]& wrapped):

python/pyarrow/fs.py

@@ -26,3 +26,20 @@ from pyarrow._fs import (  # noqa
     LocalFileSystemOptions,
     SubTreeFileSystem
 )
+
+try:
+    from pyarrow._hdfs import HdfsOptions, HadoopFileSystem  # noqa
+except ImportError:
+    pass
+
+try:
+    from pyarrow._s3fs import (  # noqa
+        initialize_s3,
+        finalize_s3,
+        S3Options,
+        S3FileSystem
+    )
+except ImportError:
+    pass
+else:
+    initialize_s3()

python/pyarrow/includes/libarrow.pxd

@@ -1018,11 +1018,11 @@ cdef extern from "arrow/io/api.h" namespace "arrow::io" nogil:
     cdef cppclass HdfsOutputStream(COutputStream):
         pass
 
-    cdef cppclass CHadoopFileSystem \
+    cdef cppclass CIOHadoopFileSystem \
             "arrow::io::HadoopFileSystem"(CIOFileSystem):
         @staticmethod
         CStatus Connect(const HdfsConnectionConfig* config,
-                        shared_ptr[CHadoopFileSystem]* client)
+                        shared_ptr[CIOHadoopFileSystem]* client)
 
         CStatus MakeDirectory(const c_string& path)
 
@@ -1077,77 +1077,6 @@ cdef extern from "arrow/io/api.h" namespace "arrow::io" nogil:
         void set_memcopy_threshold(int64_t threshold)
 
 
-cdef extern from "arrow/filesystem/api.h" namespace "arrow::fs" nogil:
-
-    ctypedef enum CFileType "arrow::fs::FileType":
-        CFileType_NonExistent "arrow::fs::FileType::NonExistent"
-        CFileType_Unknown "arrow::fs::FileType::Unknown"
-        CFileType_File "arrow::fs::FileType::File"
-        CFileType_Directory "arrow::fs::FileType::Directory"
-
-    cdef cppclass CTimePoint "arrow::fs::TimePoint":
-        pass
-
-    cdef cppclass CFileStats "arrow::fs::FileStats":
-        CFileStats()
-        CFileStats(CFileStats&&)
-        CFileStats& operator=(CFileStats&&)
-        CFileStats(const CFileStats&)
-        CFileStats& operator=(const CFileStats&)
-
-        CFileType type()
-        void set_type(CFileType type)
-        c_string path()
-        void set_path(const c_string& path)
-        c_string base_name()
-        int64_t size()
-        void set_size(int64_t size)
-        c_string extension()
-        CTimePoint mtime()
-        void set_mtime(CTimePoint mtime)
-
-    cdef cppclass CSelector "arrow::fs::Selector":
-        CSelector()
-        c_string base_dir
-        c_bool allow_non_existent
-        c_bool recursive
-
-    cdef cppclass CFileSystem "arrow::fs::FileSystem":
-        CResult[CFileStats] GetTargetStats(const c_string& path)
-        CResult[vector[CFileStats]] GetTargetStats(
-            const vector[c_string]& paths)
-        CResult[vector[CFileStats]] GetTargetStats(const CSelector& select)
-        CStatus CreateDir(const c_string& path, c_bool recursive)
-        CStatus DeleteDir(const c_string& path)
-        CStatus DeleteFile(const c_string& path)
-        CStatus DeleteFiles(const vector[c_string]& paths)
-        CStatus Move(const c_string& src, const c_string& dest)
-        CStatus CopyFile(const c_string& src, const c_string& dest)
-        CResult[shared_ptr[CInputStream]] OpenInputStream(
-            const c_string& path)
-        CResult[shared_ptr[CRandomAccessFile]] OpenInputFile(
-            const c_string& path)
-        CResult[shared_ptr[COutputStream]] OpenOutputStream(
-            const c_string& path)
-        CResult[shared_ptr[COutputStream]] OpenAppendStream(
-            const c_string& path)
-
-    cdef cppclass CLocalFileSystemOptions "arrow::fs::LocalFileSystemOptions":
-        c_bool use_mmap
-
-        @staticmethod
-        CLocalFileSystemOptions Defaults()
-
-    cdef cppclass CLocalFileSystem "arrow::fs::LocalFileSystem"(CFileSystem):
-        CLocalFileSystem()
-        CLocalFileSystem(CLocalFileSystemOptions)
-
-    cdef cppclass CSubTreeFileSystem \
-            "arrow::fs::SubTreeFileSystem"(CFileSystem):
-        CSubTreeFileSystem(const c_string& base_path,
-                           shared_ptr[CFileSystem] base_fs)
-
-
 cdef extern from "arrow/ipc/api.h" namespace "arrow::ipc" nogil:
     enum MessageType" arrow::ipc::Message::Type":
         MessageType_SCHEMA" arrow::ipc::Message::SCHEMA"

python/pyarrow/includes/libarrow_fs.pxd

@@ -0,0 +1,149 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# distutils: language = c++
+
+from libcpp.functional cimport function
+
+from pyarrow.includes.common cimport *
+from pyarrow.includes.libarrow cimport *
+
+
+cdef extern from "arrow/filesystem/api.h" namespace "arrow::fs" nogil:
+
+    ctypedef enum CFileType "arrow::fs::FileType":
+        CFileType_NonExistent "arrow::fs::FileType::NonExistent"
+        CFileType_Unknown "arrow::fs::FileType::Unknown"
+        CFileType_File "arrow::fs::FileType::File"
+        CFileType_Directory "arrow::fs::FileType::Directory"
+
+    cdef cppclass CTimePoint "arrow::fs::TimePoint":
+        pass
+
+    cdef cppclass CFileStats "arrow::fs::FileStats":
+        CFileStats()
+        CFileStats(CFileStats&&)
+        CFileStats& operator=(CFileStats&&)
+        CFileStats(const CFileStats&)
+        CFileStats& operator=(const CFileStats&)
+
+        CFileType type()
+        void set_type(CFileType type)
+        c_string path()
+        void set_path(const c_string& path)
+        c_string base_name()
+        int64_t size()
+        void set_size(int64_t size)
+        c_string extension()
+        CTimePoint mtime()
+        void set_mtime(CTimePoint mtime)
+
+    cdef cppclass CSelector "arrow::fs::Selector":
+        CSelector()
+        c_string base_dir
+        c_bool allow_non_existent
+        c_bool recursive
+
+    cdef cppclass CFileSystem "arrow::fs::FileSystem":
+        CResult[CFileStats] GetTargetStats(const c_string& path)
+        CResult[vector[CFileStats]] GetTargetStats(
+            const vector[c_string]& paths)
+        CResult[vector[CFileStats]] GetTargetStats(const CSelector& select)
+        CStatus CreateDir(const c_string& path, c_bool recursive)
+        CStatus DeleteDir(const c_string& path)
+        CStatus DeleteFile(const c_string& path)
+        CStatus DeleteFiles(const vector[c_string]& paths)
+        CStatus Move(const c_string& src, const c_string& dest)
+        CStatus CopyFile(const c_string& src, const c_string& dest)
+        CResult[shared_ptr[CInputStream]] OpenInputStream(
+            const c_string& path)
+        CResult[shared_ptr[CRandomAccessFile]] OpenInputFile(
+            const c_string& path)
+        CResult[shared_ptr[COutputStream]] OpenOutputStream(
+            const c_string& path)
+        CResult[shared_ptr[COutputStream]] OpenAppendStream(
+            const c_string& path)
+
+    cdef cppclass CLocalFileSystemOptions "arrow::fs::LocalFileSystemOptions":
+        c_bool use_mmap
+
+        @staticmethod
+        CLocalFileSystemOptions Defaults()
+
+    cdef cppclass CLocalFileSystem "arrow::fs::LocalFileSystem"(CFileSystem):
+        CLocalFileSystem()
+        CLocalFileSystem(CLocalFileSystemOptions)
+
+    cdef cppclass CSubTreeFileSystem \
+            "arrow::fs::SubTreeFileSystem"(CFileSystem):
+        CSubTreeFileSystem(const c_string& base_path,
+                           shared_ptr[CFileSystem] base_fs)
+
+    ctypedef enum CS3LogLevel "arrow::fs::S3LogLevel":
+        CS3LogLevel_Off "arrow::fs::S3LogLevel::Off"
+        CS3LogLevel_Fatal "arrow::fs::S3LogLevel::Fatal"
+        CS3LogLevel_Error "arrow::fs::S3LogLevel::Error"
+        CS3LogLevel_Warn "arrow::fs::S3LogLevel::Warn"
+        CS3LogLevel_Info "arrow::fs::S3LogLevel::Info"
+        CS3LogLevel_Debug "arrow::fs::S3LogLevel::Debug"
+        CS3LogLevel_Trace "arrow::fs::S3LogLevel::Trace"
+
+    cdef struct CS3GlobalOptions "arrow::fs::S3GlobalOptions":
+        CS3LogLevel log_level
+
+    cdef cppclass CS3Options "arrow::fs::S3Options":
+        c_string region
+        c_string endpoint_override
+        c_string scheme
+        c_bool background_writes
+        void ConfigureDefaultCredentials()
+        void ConfigureAccessKey(const c_string& access_key,
+                                const c_string& secret_key)
+
+        @staticmethod
+        CS3Options Defaults()
+        @staticmethod
+        CS3Options FromAccessKey(const c_string& access_key,
+                                 const c_string& secret_key)
+
+    cdef cppclass CS3FileSystem "arrow::fs::S3FileSystem"(CFileSystem):
+        @staticmethod
+        CResult[shared_ptr[CS3FileSystem]] Make(const CS3Options& options)
+
+    cdef CStatus CInitializeS3 "arrow::fs::InitializeS3"(
+        const CS3GlobalOptions& options)
+    cdef CStatus CFinalizeS3 "arrow::fs::FinalizeS3"()
+
+    cdef cppclass CHdfsOptions "arrow::fs::HdfsOptions":
+        HdfsConnectionConfig connection_config
+        int32_t buffer_size
+        int16_t replication
+        int64_t default_block_size
+        @staticmethod
+        CResult[CHdfsOptions] FromUriString "FromUri"(
+            const c_string& uri_string)
+        void ConfigureEndPoint(const c_string& host, int port)
+        void ConfigureHdfsDriver(c_bool use_hdfs3)
+        void ConfigureHdfsReplication(int16_t replication)
+        void ConfigureHdfsUser(const c_string& user_name)
+        void ConfigureHdfsBufferSize(int32_t buffer_size)
+        void ConfigureHdfsBlockSize(int64_t default_block_size)
+
+    cdef cppclass CHadoopFileSystem "arrow::fs::HadoopFileSystem"(CFileSystem):
+        @staticmethod
+        CResult[shared_ptr[CHadoopFileSystem]] Make(
+            const CHdfsOptions& options)

python/pyarrow/includes/libarrow_s3fs.pxd

@@ -1,60 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements.  See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership.  The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License.  You may obtain a copy of the License at
-#
-#   http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied.  See the License for the
-# specific language governing permissions and limitations
-# under the License.
-
-# distutils: language = c++
-
-from libcpp.functional cimport function
-
-from pyarrow.includes.common cimport *
-from pyarrow.includes.libarrow cimport CFileSystem
-
-cdef extern from "arrow/filesystem/api.h" namespace "arrow::fs" nogil:
-
-    ctypedef enum CS3LogLevel "arrow::fs::S3LogLevel":
-        CS3LogLevel_Off "arrow::fs::S3LogLevel::Off"
-        CS3LogLevel_Fatal "arrow::fs::S3LogLevel::Fatal"
-        CS3LogLevel_Error "arrow::fs::S3LogLevel::Error"
-        CS3LogLevel_Warn "arrow::fs::S3LogLevel::Warn"
-        CS3LogLevel_Info "arrow::fs::S3LogLevel::Info"
-        CS3LogLevel_Debug "arrow::fs::S3LogLevel::Debug"
-        CS3LogLevel_Trace "arrow::fs::S3LogLevel::Trace"
-
-    cdef struct CS3GlobalOptions "arrow::fs::S3GlobalOptions":
-        CS3LogLevel log_level
-
-    cdef cppclass CS3Options "arrow::fs::S3Options":
-        c_string region
-        c_string endpoint_override
-        c_string scheme
-        c_bool background_writes
-        void ConfigureDefaultCredentials()
-        void ConfigureAccessKey(const c_string& access_key,
-                                const c_string& secret_key)
-
-        @staticmethod
-        CS3Options Defaults()
-        @staticmethod
-        CS3Options FromAccessKey(const c_string& access_key,
-                                 const c_string& secret_key)
-
-    cdef cppclass CS3FileSystem "arrow::fs::S3FileSystem"(CFileSystem):
-        @staticmethod
-        CResult[shared_ptr[CS3FileSystem]] Make(const CS3Options& options)
-
-    cdef CStatus CInitializeS3 "arrow::fs::InitializeS3"(
-        const CS3GlobalOptions& options)
-    cdef CStatus CFinalizeS3 "arrow::fs::FinalizeS3"()

python/pyarrow/io-hdfs.pxi

@@ -55,7 +55,7 @@ def strip_hdfs_abspath(path):
 
 cdef class HadoopFileSystem:
     cdef:
-        shared_ptr[CHadoopFileSystem] client
+        shared_ptr[CIOHadoopFileSystem] client
 
     cdef readonly:
         bint is_open
@@ -102,7 +102,7 @@ cdef class HadoopFileSystem:
         self.extra_conf = extra_conf
 
         with nogil:
-            check_status(CHadoopFileSystem.Connect(&conf, &self.client))
+            check_status(CIOHadoopFileSystem.Connect(&conf, &self.client))
         self.is_open = True
 
     @classmethod

python/pyarrow/tests/conftest.py

@@ -118,7 +118,7 @@ except ImportError:
     pass
 
 try:
-    import pyarrow.plasma as plasma  # noqa
+    import pyarrow.plasma  # noqa
     defaults['plasma'] = True
 except ImportError:
     pass
@@ -136,11 +136,17 @@ except ImportError:
     pass
 
 try:
-    import pyarrow.s3fs  # noqa
+    import pyarrow.fs import S3FileSystem  # noqa
     defaults['s3'] = True
 except ImportError:
     pass
 
+try:
+    from pyarrow.fs import HadoopFileSystem   # noqa
+    defaults['hdfs'] = True
+except ImportError:
+    pass
+
 
 def pytest_addoption(parser):
     # Create options to selectively enable test groups
@@ -230,6 +236,19 @@ except ImportError:
             shutil.rmtree(self.tmp)
 
 
+# TODO(kszucs): move the following fixtures to test_fs.py once the previous
+# parquet dataset implementation and hdfs implementation are removed.
+
+@pytest.mark.hdfs
+@pytest.fixture(scope='session')
+def hdfs_server():
+    host = os.environ.get('ARROW_HDFS_TEST_HOST', 'default')
+    port = int(os.environ.get('ARROW_HDFS_TEST_PORT', 0))
+    user = os.environ.get('ARROW_HDFS_TEST_USER', 'hdfs')
+    uri = "hdfs://{}:{}".format(host, port)
+    return uri, user
+
+
 @pytest.mark.s3
 @pytest.fixture(scope='session')
 def minio_server():

python/pyarrow/tests/test_fs.py

@@ -64,7 +64,7 @@ def subtree_localfs(request, tempdir, localfs):
 @pytest.fixture
 def s3fs(request, minio_server):
     request.config.pyarrow.requires('s3')
-    from pyarrow.s3fs import S3Options, S3FileSystem
+    from pyarrow.fs import S3Options, S3FileSystem
 
     address, access_key, secret_key = minio_server
     bucket = 'pyarrow-filesystem/'
@@ -96,6 +96,25 @@ def subtree_s3fs(request, s3fs):
     )
 
 
+@pytest.fixture
+def hdfs(request, hdfs_server):
+    request.config.pyarrow.requires('hdfs')
+    from pyarrow.fs import HdfsOptions, HadoopFileSystem
+
+    uri, user = hdfs_server
+    options = HdfsOptions.from_uri(uri)
+    options.user = user
+
+    fs = HadoopFileSystem(options)
+
+    return dict(
+        fs=fs,
+        pathfn=lambda p: p,
+        allow_move_dir=True,
+        allow_append_to_file=False,
+    )
+
+
 @pytest.fixture(params=[
     pytest.param(
         pytest.lazy_fixture('localfs'),
@@ -113,6 +132,10 @@ def subtree_s3fs(request, s3fs):
         pytest.lazy_fixture('s3fs'),
         id='S3FileSystem'
     ),
+    pytest.param(
+        pytest.lazy_fixture('hdfs'),
+        id='HadoopFileSystem'
+    ),
 ])
 def filesystem_config(request):
     return request.param
@@ -423,7 +446,7 @@ def test_localfs_options():
 
 @pytest.mark.s3
 def test_s3_options(minio_server):
-    from pyarrow.s3fs import S3Options
+    from pyarrow.fs import S3Options
 
     options = S3Options()
 

python/setup.py

@@ -143,6 +143,8 @@ class build_ext(_build_ext):
 
         self.with_s3 = strtobool(
             os.environ.get('PYARROW_WITH_S3', '0'))
+        self.with_hdfs = strtobool(
+            os.environ.get('PYARROW_WITH_HDFS', '0'))
         self.with_cuda = strtobool(
             os.environ.get('PYARROW_WITH_CUDA', '0'))
         self.with_flight = strtobool(
@@ -179,6 +181,7 @@ class build_ext(_build_ext):
         '_orc',
         '_plasma',
         '_s3fs',
+        '_hdfs',
         'gandiva']
 
     def _run_cmake(self):
@@ -231,6 +234,7 @@ class build_ext(_build_ext):
             append_cmake_bool(self.with_parquet, 'PYARROW_BUILD_PARQUET')
             append_cmake_bool(self.with_plasma, 'PYARROW_BUILD_PLASMA')
             append_cmake_bool(self.with_s3, 'PYARROW_BUILD_S3')
+            append_cmake_bool(self.with_hdfs, 'PYARROW_BUILD_HDFS')
             append_cmake_bool(self.with_tensorflow, 'PYARROW_USE_TENSORFLOW')
             append_cmake_bool(self.bundle_arrow_cpp,
                               'PYARROW_BUNDLE_ARROW_CPP')
@@ -402,6 +406,8 @@ class build_ext(_build_ext):
             return True
         if name == '_s3fs' and not self.with_s3:
             return True
+        if name == '_hdfs' and not self.with_hdfs:
+            return True
         if name == '_cuda' and not self.with_cuda:
             return True
         if name == 'gandiva' and not self.with_gandiva: