ARROW-2661: [Python] Adding the ability to programmatically pass hdfs configration key/value pairs via pyarrow

https://issues.apache.org/jira/browse/ARROW-2661

Both the JNI and libhdfs3 support hdfsBuilderConfSetStr so we can utilize that to allow passing arbitrary configuration values for hdfs connection similiar to how https://hdfs3.readthedocs.io/en/latest/hdfs.html supports passing them.

I've added a param called `extra_conf` to facilitate it in pyarrow, such as:

```python
import pyarrow
conf = {"dfs.nameservices": "nameservice1",
        "dfs.ha.namenodes.nameservice1": "namenode113,namenode188",
        "dfs.namenode.rpc-address.nameservice1.namenode113": "hostname_of_server1:8020",
        "dfs.namenode.rpc-address.nameservice1.namenode188": "hostname_of_server2:8020",
        "dfs.namenode.http-address.nameservice1.namenode188": "hostname_of_server1:50070",
        "dfs.namenode.http-address.nameservice1.namenode188": "hostname_of_server2:50070",
        "hadoop.security.authentication": "kerberos"
}
hdfs = pyarrow.hdfs.connect(host='nameservice1', driver='libhdfs3', extra_conf=conf)
```

Author: Matthew Topol <mtopol@factset.com>

Closes #2097 from zeroshade/configs and squashes the following commits:

047dd4b <Matthew Topol> forgot to use make format to fix the order of includes. oops
d27e3c3 <Matthew Topol> switching to unordered_map
858b44b <Matthew Topol> missed a flake8 spot
77eeae0 <Matthew Topol>  Adding the ability to programmatically pass hdfs configuration key/value pairs in the C++ and via pyarrow

Author: Matthew Topol

cpp/src/arrow/io/hdfs-internal.cc

@@ -318,6 +318,10 @@ hdfsFS LibHdfsShim::BuilderConnect(hdfsBuilder* bld) {
   return this->hdfsBuilderConnect(bld);
 }
 
+int LibHdfsShim::BuilderConfSetStr(hdfsBuilder* bld, const char* key, const char* val) {
+  return this->hdfsBuilderConfSetStr(bld, key, val);
+}
+
 int LibHdfsShim::Disconnect(hdfsFS fs) { return this->hdfsDisconnect(fs); }
 
 hdfsFile LibHdfsShim::OpenFile(hdfsFS fs, const char* path, int flags, int bufferSize,
@@ -495,6 +499,7 @@ Status LibHdfsShim::GetRequiredSymbols() {
   GET_SYMBOL_REQUIRED(this, hdfsBuilderSetUserName);
   GET_SYMBOL_REQUIRED(this, hdfsBuilderSetKerbTicketCachePath);
   GET_SYMBOL_REQUIRED(this, hdfsBuilderSetForceNewInstance);
+  GET_SYMBOL_REQUIRED(this, hdfsBuilderConfSetStr);
   GET_SYMBOL_REQUIRED(this, hdfsBuilderConnect);
   GET_SYMBOL_REQUIRED(this, hdfsCreateDirectory);
   GET_SYMBOL_REQUIRED(this, hdfsDelete);

cpp/src/arrow/io/hdfs-internal.h

@@ -53,6 +53,7 @@ struct LibHdfsShim {
                                             const char* kerbTicketCachePath);
   void (*hdfsBuilderSetForceNewInstance)(hdfsBuilder* bld);
   hdfsFS (*hdfsBuilderConnect)(hdfsBuilder* bld);
+  int (*hdfsBuilderConfSetStr)(hdfsBuilder* bld, const char* key, const char* val);
 
   int (*hdfsDisconnect)(hdfsFS fs);
 
@@ -97,6 +98,7 @@ struct LibHdfsShim {
     this->hdfsBuilderSetUserName = nullptr;
     this->hdfsBuilderSetKerbTicketCachePath = nullptr;
     this->hdfsBuilderSetForceNewInstance = nullptr;
+    this->hdfsBuilderConfSetStr = nullptr;
     this->hdfsBuilderConnect = nullptr;
     this->hdfsDisconnect = nullptr;
     this->hdfsOpenFile = nullptr;
@@ -142,6 +144,8 @@ struct LibHdfsShim {
 
   void BuilderSetForceNewInstance(hdfsBuilder* bld);
 
+  int BuilderConfSetStr(hdfsBuilder* bld, const char* key, const char* val);
+
   hdfsFS BuilderConnect(hdfsBuilder* bld);
 
   int Disconnect(hdfsFS fs);

cpp/src/arrow/io/hdfs.cc

@@ -335,6 +335,12 @@ class HadoopFileSystem::HadoopFileSystemImpl {
     if (!config->kerb_ticket.empty()) {
       driver_->BuilderSetKerbTicketCachePath(builder, config->kerb_ticket.c_str());
     }
+
+    for (auto& kv : config->extra_conf) {
+      int ret = driver_->BuilderConfSetStr(builder, kv.first.c_str(), kv.second.c_str());
+      CHECK_FAILURE(ret, "confsetstr");
+    }
+
     driver_->BuilderSetForceNewInstance(builder);
     fs_ = driver_->BuilderConnect(builder);
 

cpp/src/arrow/io/hdfs.h

@@ -21,6 +21,7 @@
 #include <cstdint>
 #include <memory>
 #include <string>
+#include <unordered_map>
 #include <vector>
 
 #include "arrow/io/interfaces.h"
@@ -63,6 +64,7 @@ struct HdfsConnectionConfig {
   int port;
   std::string user;
   std::string kerb_ticket;
+  std::unordered_map<std::string, std::string> extra_conf;
   HdfsDriver driver;
 };
 

python/pyarrow/hdfs.py

@@ -30,15 +30,16 @@ class HadoopFileSystem(lib.HadoopFileSystem, FileSystem):
     """
 
     def __init__(self, host="default", port=0, user=None, kerb_ticket=None,
-                 driver='libhdfs'):
+                 driver='libhdfs', extra_conf=None):
         if driver == 'libhdfs':
             _maybe_set_hadoop_classpath()
 
-        self._connect(host, port, user, kerb_ticket, driver)
+        self._connect(host, port, user, kerb_ticket, driver, extra_conf)
 
     def __reduce__(self):
         return (HadoopFileSystem, (self.host, self.port, self.user,
-                                   self.kerb_ticket, self.driver))
+                                   self.kerb_ticket, self.driver,
+                                   self.extra_conf))
 
     def _isfilestore(self):
         """
@@ -149,7 +150,7 @@ def _libhdfs_walk_files_dirs(top_path, contents):
 
 
 def connect(host="default", port=0, user=None, kerb_ticket=None,
-            driver='libhdfs'):
+            driver='libhdfs', extra_conf=None):
     """
     Connect to an HDFS cluster. All parameters are optional and should
     only be set if the defaults need to be overridden.
@@ -178,5 +179,6 @@ def connect(host="default", port=0, user=None, kerb_ticket=None,
     filesystem : HadoopFileSystem
     """
     fs = HadoopFileSystem(host=host, port=port, user=user,
-                          kerb_ticket=kerb_ticket, driver=driver)
+                          kerb_ticket=kerb_ticket, driver=driver,
+                          extra_conf=extra_conf)
     return fs

python/pyarrow/includes/libarrow.pxd

@@ -19,7 +19,6 @@
 
 from pyarrow.includes.common cimport *
 
-
 cdef extern from "arrow/util/key_value_metadata.h" namespace "arrow" nogil:
     cdef cppclass CKeyValueMetadata" arrow::KeyValueMetadata":
         CKeyValueMetadata()
@@ -656,6 +655,7 @@ cdef extern from "arrow/io/api.h" namespace "arrow::io" nogil:
         int port
         c_string user
         c_string kerb_ticket
+        unordered_map[c_string, c_string] extra_conf
         HdfsDriver driver
 
     cdef cppclass HdfsPathInfo:

python/pyarrow/io-hdfs.pxi

@@ -64,8 +64,9 @@ cdef class HadoopFileSystem:
         str kerb_ticket
         str driver
         int port
+        dict extra_conf
 
-    def _connect(self, host, port, user, kerb_ticket, driver):
+    def _connect(self, host, port, user, kerb_ticket, driver, extra_conf):
         cdef HdfsConnectionConfig conf
 
         if host is not None:
@@ -95,6 +96,11 @@ cdef class HadoopFileSystem:
             raise ValueError("unknown driver: %r" % driver)
         self.driver = driver
 
+        if extra_conf is not None and isinstance(extra_conf, dict):
+            conf.extra_conf = {tobytes(k): tobytes(v)
+                               for k, v in extra_conf.items()}
+        self.extra_conf = extra_conf
+
         with nogil:
             check_status(CHadoopFileSystem.Connect(&conf, &self.client))
         self.is_open = True