You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@arrow.apache.org by uw...@apache.org on 2017/08/01 13:51:25 UTC
arrow git commit: ARROW-1301: [C++/Python] More complete filesystem
API for HDFS
Repository: arrow
Updated Branches:
refs/heads/master 3a84653a3 -> e1d574c7c
ARROW-1301: [C++/Python] More complete filesystem API for HDFS
This also includes a fair bit of API normalization and cleaning.
Author: Wes McKinney <we...@twosigma.com>
Closes #926 from wesm/ARROW-1301 and squashes the following commits:
bcc9310 [Wes McKinney] Add missing API
8bf51f5 [Wes McKinney] Add more filesystem methods, tests for HDFS
98847b5 [Wes McKinney] Some HDFS refactoring. Implement chmod, chown. Normalize Filesystem->FileSystem
Project: http://git-wip-us.apache.org/repos/asf/arrow/repo
Commit: http://git-wip-us.apache.org/repos/asf/arrow/commit/e1d574c7
Tree: http://git-wip-us.apache.org/repos/asf/arrow/tree/e1d574c7
Diff: http://git-wip-us.apache.org/repos/asf/arrow/diff/e1d574c7
Branch: refs/heads/master
Commit: e1d574c7cf2207ddde64273cd34ba8352bc003bc
Parents: 3a84653
Author: Wes McKinney <we...@twosigma.com>
Authored: Tue Aug 1 15:51:18 2017 +0200
Committer: Uwe L. Korn <uw...@xhochy.com>
Committed: Tue Aug 1 15:51:18 2017 +0200
----------------------------------------------------------------------
cpp/src/arrow/io/hdfs-internal.cc | 16 +-
cpp/src/arrow/io/hdfs.cc | 108 +++++--
cpp/src/arrow/io/hdfs.h | 48 ++-
cpp/src/arrow/io/interfaces.h | 25 +-
cpp/src/arrow/io/io-hdfs-test.cc | 59 ++--
python/doc/source/api.rst | 10 +-
python/doc/source/conf.py | 3 +-
python/doc/source/filesystems.rst | 43 ++-
python/doc/source/memory.rst | 7 -
python/pyarrow/__init__.py | 6 +-
python/pyarrow/filesystem.py | 105 +++++--
python/pyarrow/hdfs.py | 34 ++-
python/pyarrow/includes/libarrow.pxd | 15 +-
python/pyarrow/io-hdfs.pxi | 468 ++++++++++++++++++++++++++++++
python/pyarrow/io.pxi | 298 -------------------
python/pyarrow/lib.pyx | 1 +
python/pyarrow/parquet.py | 14 +-
python/pyarrow/tests/test_hdfs.py | 97 ++++++-
python/pyarrow/tests/test_parquet.py | 4 +-
19 files changed, 915 insertions(+), 446 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/arrow/blob/e1d574c7/cpp/src/arrow/io/hdfs-internal.cc
----------------------------------------------------------------------
diff --git a/cpp/src/arrow/io/hdfs-internal.cc b/cpp/src/arrow/io/hdfs-internal.cc
index 8f42b1c..fd7417b 100644
--- a/cpp/src/arrow/io/hdfs-internal.cc
+++ b/cpp/src/arrow/io/hdfs-internal.cc
@@ -462,21 +462,11 @@ tOffset LibHdfsShim::GetUsed(hdfsFS fs) { return this->hdfsGetUsed(fs); }
int LibHdfsShim::Chown(hdfsFS fs, const char* path, const char* owner,
const char* group) {
- GET_SYMBOL(this, hdfsChown);
- if (this->hdfsChown) {
- return this->hdfsChown(fs, path, owner, group);
- } else {
- return 0;
- }
+ return this->hdfsChown(fs, path, owner, group);
}
int LibHdfsShim::Chmod(hdfsFS fs, const char* path, short mode) { // NOLINT
- GET_SYMBOL(this, hdfsChmod);
- if (this->hdfsChmod) {
- return this->hdfsChmod(fs, path, mode);
- } else {
- return 0;
- }
+ return this->hdfsChmod(fs, path, mode);
}
int LibHdfsShim::Utime(hdfsFS fs, const char* path, tTime mtime, tTime atime) {
@@ -504,6 +494,8 @@ Status LibHdfsShim::GetRequiredSymbols() {
GET_SYMBOL_REQUIRED(this, hdfsGetUsed);
GET_SYMBOL_REQUIRED(this, hdfsGetPathInfo);
GET_SYMBOL_REQUIRED(this, hdfsListDirectory);
+ GET_SYMBOL_REQUIRED(this, hdfsChown);
+ GET_SYMBOL_REQUIRED(this, hdfsChmod);
// File methods
GET_SYMBOL_REQUIRED(this, hdfsCloseFile);
http://git-wip-us.apache.org/repos/asf/arrow/blob/e1d574c7/cpp/src/arrow/io/hdfs.cc
----------------------------------------------------------------------
diff --git a/cpp/src/arrow/io/hdfs.cc b/cpp/src/arrow/io/hdfs.cc
index 500f42c..254e483 100644
--- a/cpp/src/arrow/io/hdfs.cc
+++ b/cpp/src/arrow/io/hdfs.cc
@@ -308,9 +308,9 @@ static void SetPathInfo(const hdfsFileInfo* input, HdfsPathInfo* out) {
}
// Private implementation
-class HdfsClient::HdfsClientImpl {
+class HadoopFileSystem::HadoopFileSystemImpl {
public:
- HdfsClientImpl() {}
+ HadoopFileSystemImpl() {}
Status Connect(const HdfsConnectionConfig* config) {
if (config->driver == HdfsDriver::LIBHDFS3) {
@@ -396,6 +396,24 @@ class HdfsClient::HdfsClientImpl {
return Status::OK();
}
+ Status Stat(const std::string& path, FileStatistics* stat) {
+ HdfsPathInfo info;
+ RETURN_NOT_OK(GetPathInfo(path, &info));
+
+ stat->size = info.size;
+ stat->kind = info.kind;
+ return Status::OK();
+ }
+
+ Status GetChildren(const std::string& path, std::vector<std::string>* listing) {
+ std::vector<HdfsPathInfo> detailed_listing;
+ RETURN_NOT_OK(ListDirectory(path, &detailed_listing));
+ for (const auto& info : detailed_listing) {
+ listing->push_back(info.name);
+ }
+ return Status::OK();
+ }
+
Status ListDirectory(const std::string& path, std::vector<HdfsPathInfo>* listing) {
int num_entries = 0;
hdfsFileInfo* entries = driver_->ListDirectory(fs_, path.c_str(), &num_entries);
@@ -476,6 +494,18 @@ class HdfsClient::HdfsClientImpl {
return Status::OK();
}
+ Status Chmod(const std::string& path, int mode) {
+ int ret = driver_->Chmod(fs_, path.c_str(), static_cast<short>(mode)); // NOLINT
+ CHECK_FAILURE(ret, "Chmod");
+ return Status::OK();
+ }
+
+ Status Chown(const std::string& path, const char* owner, const char* group) {
+ int ret = driver_->Chown(fs_, path.c_str(), owner, group);
+ CHECK_FAILURE(ret, "Chown");
+ return Status::OK();
+ }
+
private:
LibHdfsShim* driver_;
@@ -490,68 +520,92 @@ class HdfsClient::HdfsClientImpl {
// ----------------------------------------------------------------------
// Public API for HDFSClient
-HdfsClient::HdfsClient() { impl_.reset(new HdfsClientImpl()); }
+HadoopFileSystem::HadoopFileSystem() { impl_.reset(new HadoopFileSystemImpl()); }
-HdfsClient::~HdfsClient() {}
+HadoopFileSystem::~HadoopFileSystem() {}
-Status HdfsClient::Connect(const HdfsConnectionConfig* config,
- std::shared_ptr<HdfsClient>* fs) {
+Status HadoopFileSystem::Connect(const HdfsConnectionConfig* config,
+ std::shared_ptr<HadoopFileSystem>* fs) {
// ctor is private, make_shared will not work
- *fs = std::shared_ptr<HdfsClient>(new HdfsClient());
+ *fs = std::shared_ptr<HadoopFileSystem>(new HadoopFileSystem());
RETURN_NOT_OK((*fs)->impl_->Connect(config));
return Status::OK();
}
-Status HdfsClient::MakeDirectory(const std::string& path) {
+Status HadoopFileSystem::MakeDirectory(const std::string& path) {
return impl_->MakeDirectory(path);
}
-Status HdfsClient::Delete(const std::string& path, bool recursive) {
+Status HadoopFileSystem::Delete(const std::string& path, bool recursive) {
return impl_->Delete(path, recursive);
}
-Status HdfsClient::Disconnect() { return impl_->Disconnect(); }
+Status HadoopFileSystem::DeleteDirectory(const std::string& path) {
+ return Delete(path, true);
+}
+
+Status HadoopFileSystem::Disconnect() { return impl_->Disconnect(); }
-bool HdfsClient::Exists(const std::string& path) { return impl_->Exists(path); }
+bool HadoopFileSystem::Exists(const std::string& path) { return impl_->Exists(path); }
-Status HdfsClient::GetPathInfo(const std::string& path, HdfsPathInfo* info) {
+Status HadoopFileSystem::GetPathInfo(const std::string& path, HdfsPathInfo* info) {
return impl_->GetPathInfo(path, info);
}
-Status HdfsClient::GetCapacity(int64_t* nbytes) { return impl_->GetCapacity(nbytes); }
+Status HadoopFileSystem::Stat(const std::string& path, FileStatistics* stat) {
+ return impl_->Stat(path, stat);
+}
+
+Status HadoopFileSystem::GetCapacity(int64_t* nbytes) {
+ return impl_->GetCapacity(nbytes);
+}
+
+Status HadoopFileSystem::GetUsed(int64_t* nbytes) { return impl_->GetUsed(nbytes); }
-Status HdfsClient::GetUsed(int64_t* nbytes) { return impl_->GetUsed(nbytes); }
+Status HadoopFileSystem::GetChildren(const std::string& path,
+ std::vector<std::string>* listing) {
+ return impl_->GetChildren(path, listing);
+}
-Status HdfsClient::ListDirectory(const std::string& path,
- std::vector<HdfsPathInfo>* listing) {
+Status HadoopFileSystem::ListDirectory(const std::string& path,
+ std::vector<HdfsPathInfo>* listing) {
return impl_->ListDirectory(path, listing);
}
-Status HdfsClient::OpenReadable(const std::string& path, int32_t buffer_size,
- std::shared_ptr<HdfsReadableFile>* file) {
+Status HadoopFileSystem::OpenReadable(const std::string& path, int32_t buffer_size,
+ std::shared_ptr<HdfsReadableFile>* file) {
return impl_->OpenReadable(path, buffer_size, file);
}
-Status HdfsClient::OpenReadable(const std::string& path,
- std::shared_ptr<HdfsReadableFile>* file) {
+Status HadoopFileSystem::OpenReadable(const std::string& path,
+ std::shared_ptr<HdfsReadableFile>* file) {
return OpenReadable(path, kDefaultHdfsBufferSize, file);
}
-Status HdfsClient::OpenWriteable(const std::string& path, bool append,
- int32_t buffer_size, int16_t replication,
- int64_t default_block_size,
- std::shared_ptr<HdfsOutputStream>* file) {
+Status HadoopFileSystem::OpenWriteable(const std::string& path, bool append,
+ int32_t buffer_size, int16_t replication,
+ int64_t default_block_size,
+ std::shared_ptr<HdfsOutputStream>* file) {
return impl_->OpenWriteable(path, append, buffer_size, replication, default_block_size,
file);
}
-Status HdfsClient::OpenWriteable(const std::string& path, bool append,
- std::shared_ptr<HdfsOutputStream>* file) {
+Status HadoopFileSystem::OpenWriteable(const std::string& path, bool append,
+ std::shared_ptr<HdfsOutputStream>* file) {
return OpenWriteable(path, append, 0, 0, 0, file);
}
-Status HdfsClient::Rename(const std::string& src, const std::string& dst) {
+Status HadoopFileSystem::Chmod(const std::string& path, int mode) {
+ return impl_->Chmod(path, mode);
+}
+
+Status HadoopFileSystem::Chown(const std::string& path, const char* owner,
+ const char* group) {
+ return impl_->Chown(path, owner, group);
+}
+
+Status HadoopFileSystem::Rename(const std::string& src, const std::string& dst) {
return impl_->Rename(src, dst);
}
http://git-wip-us.apache.org/repos/asf/arrow/blob/e1d574c7/cpp/src/arrow/io/hdfs.h
----------------------------------------------------------------------
diff --git a/cpp/src/arrow/io/hdfs.h b/cpp/src/arrow/io/hdfs.h
index 63c3ae0..1507ca9 100644
--- a/cpp/src/arrow/io/hdfs.h
+++ b/cpp/src/arrow/io/hdfs.h
@@ -34,7 +34,7 @@ class Status;
namespace io {
-class HdfsClient;
+class HadoopFileSystem;
class HdfsReadableFile;
class HdfsOutputStream;
@@ -66,9 +66,9 @@ struct HdfsConnectionConfig {
HdfsDriver driver;
};
-class ARROW_EXPORT HdfsClient : public FileSystemClient {
+class ARROW_EXPORT HadoopFileSystem : public FileSystem {
public:
- ~HdfsClient();
+ ~HadoopFileSystem();
// Connect to an HDFS cluster given a configuration
//
@@ -76,13 +76,13 @@ class ARROW_EXPORT HdfsClient : public FileSystemClient {
// @param fs (out): the created client
// @returns Status
static Status Connect(const HdfsConnectionConfig* config,
- std::shared_ptr<HdfsClient>* fs);
+ std::shared_ptr<HadoopFileSystem>* fs);
// Create directory and all parents
//
// @param path (in): absolute HDFS path
// @returns Status
- Status MakeDirectory(const std::string& path);
+ Status MakeDirectory(const std::string& path) override;
// Delete file or directory
// @param path: absolute path to data
@@ -90,6 +90,8 @@ class ARROW_EXPORT HdfsClient : public FileSystemClient {
// @returns error status on failure
Status Delete(const std::string& path, bool recursive = false);
+ Status DeleteDirectory(const std::string& path) override;
+
// Disconnect from cluster
//
// @returns Status
@@ -112,18 +114,29 @@ class ARROW_EXPORT HdfsClient : public FileSystemClient {
// @returns Status
Status GetUsed(int64_t* nbytes);
+ Status GetChildren(const std::string& path, std::vector<std::string>* listing) override;
+
Status ListDirectory(const std::string& path, std::vector<HdfsPathInfo>* listing);
- // @param path file path to change
- // @param owner pass nullptr for no change
- // @param group pass nullptr for no change
+ /// Change
+ ///
+ /// @param path file path to change
+ /// @param owner pass nullptr for no change
+ /// @param group pass nullptr for no change
Status Chown(const std::string& path, const char* owner, const char* group);
+ /// Change path permissions
+ ///
+ /// \param path Absolute path in file system
+ /// \param mode Mode bitset
+ /// \return Status
Status Chmod(const std::string& path, int mode);
// Move file or directory from source path to destination path within the
// current filesystem
- Status Rename(const std::string& src, const std::string& dst);
+ Status Rename(const std::string& src, const std::string& dst) override;
+
+ Status Stat(const std::string& path, FileStatistics* stat) override;
// TODO(wesm): GetWorkingDirectory, SetWorkingDirectory
@@ -152,13 +165,18 @@ class ARROW_EXPORT HdfsClient : public FileSystemClient {
friend class HdfsReadableFile;
friend class HdfsOutputStream;
- class ARROW_NO_EXPORT HdfsClientImpl;
- std::unique_ptr<HdfsClientImpl> impl_;
+ class ARROW_NO_EXPORT HadoopFileSystemImpl;
+ std::unique_ptr<HadoopFileSystemImpl> impl_;
- HdfsClient();
- DISALLOW_COPY_AND_ASSIGN(HdfsClient);
+ HadoopFileSystem();
+ DISALLOW_COPY_AND_ASSIGN(HadoopFileSystem);
};
+// 0.6.0
+#ifndef ARROW_NO_DEPRECATED_API
+using HdfsClient = HadoopFileSystem;
+#endif
+
class ARROW_EXPORT HdfsReadableFile : public RandomAccessFile {
public:
~HdfsReadableFile();
@@ -191,7 +209,7 @@ class ARROW_EXPORT HdfsReadableFile : public RandomAccessFile {
class ARROW_NO_EXPORT HdfsReadableFileImpl;
std::unique_ptr<HdfsReadableFileImpl> impl_;
- friend class HdfsClient::HdfsClientImpl;
+ friend class HadoopFileSystem::HadoopFileSystemImpl;
DISALLOW_COPY_AND_ASSIGN(HdfsReadableFile);
};
@@ -216,7 +234,7 @@ class ARROW_EXPORT HdfsOutputStream : public OutputStream {
class ARROW_NO_EXPORT HdfsOutputStreamImpl;
std::unique_ptr<HdfsOutputStreamImpl> impl_;
- friend class HdfsClient::HdfsClientImpl;
+ friend class HadoopFileSystem::HadoopFileSystemImpl;
HdfsOutputStream();
http://git-wip-us.apache.org/repos/asf/arrow/blob/e1d574c7/cpp/src/arrow/io/interfaces.h
----------------------------------------------------------------------
diff --git a/cpp/src/arrow/io/interfaces.h b/cpp/src/arrow/io/interfaces.h
index e71a5c9..4bb7ebe 100644
--- a/cpp/src/arrow/io/interfaces.h
+++ b/cpp/src/arrow/io/interfaces.h
@@ -22,6 +22,7 @@
#include <memory>
#include <mutex>
#include <string>
+#include <vector>
#include "arrow/util/macros.h"
#include "arrow/util/visibility.h"
@@ -42,9 +43,29 @@ struct ObjectType {
enum type { FILE, DIRECTORY };
};
-class ARROW_EXPORT FileSystemClient {
+struct ARROW_EXPORT FileStatistics {
+ /// Size of file, -1 if finding length is unsupported
+ int64_t size;
+ ObjectType::type kind;
+
+ FileStatistics() {}
+ FileStatistics(int64_t size, ObjectType::type kind) : size(size), kind(kind) {}
+};
+
+class ARROW_EXPORT FileSystem {
public:
- virtual ~FileSystemClient() {}
+ virtual ~FileSystem() {}
+
+ virtual Status MakeDirectory(const std::string& path) = 0;
+
+ virtual Status DeleteDirectory(const std::string& path) = 0;
+
+ virtual Status GetChildren(const std::string& path,
+ std::vector<std::string>* listing) = 0;
+
+ virtual Status Rename(const std::string& src, const std::string& dst) = 0;
+
+ virtual Status Stat(const std::string& path, FileStatistics* stat) = 0;
};
class ARROW_EXPORT FileInterface {
http://git-wip-us.apache.org/repos/asf/arrow/blob/e1d574c7/cpp/src/arrow/io/io-hdfs-test.cc
----------------------------------------------------------------------
diff --git a/cpp/src/arrow/io/io-hdfs-test.cc b/cpp/src/arrow/io/io-hdfs-test.cc
index c584cf5..92f4291 100644
--- a/cpp/src/arrow/io/io-hdfs-test.cc
+++ b/cpp/src/arrow/io/io-hdfs-test.cc
@@ -48,7 +48,7 @@ struct PivotalDriver {
};
template <typename DRIVER>
-class TestHdfsClient : public ::testing::Test {
+class TestHadoopFileSystem : public ::testing::Test {
public:
Status MakeScratchDir() {
if (client_->Exists(scratch_dir_)) {
@@ -124,7 +124,7 @@ class TestHdfsClient : public ::testing::Test {
conf_.port = port == nullptr ? 20500 : atoi(port);
conf_.driver = DRIVER::type;
- ASSERT_OK(HdfsClient::Connect(&conf_, &client_));
+ ASSERT_OK(HadoopFileSystem::Connect(&conf_, &client_));
}
void TearDown() {
@@ -141,11 +141,11 @@ class TestHdfsClient : public ::testing::Test {
// Resources shared amongst unit tests
std::string scratch_dir_;
- std::shared_ptr<HdfsClient> client_;
+ std::shared_ptr<HadoopFileSystem> client_;
};
template <>
-std::string TestHdfsClient<PivotalDriver>::HdfsAbsPath(const std::string& relpath) {
+std::string TestHadoopFileSystem<PivotalDriver>::HdfsAbsPath(const std::string& relpath) {
std::stringstream ss;
ss << relpath;
return ss.str();
@@ -161,17 +161,17 @@ HdfsDriver JNIDriver::type = HdfsDriver::LIBHDFS;
HdfsDriver PivotalDriver::type = HdfsDriver::LIBHDFS3;
typedef ::testing::Types<JNIDriver, PivotalDriver> DriverTypes;
-TYPED_TEST_CASE(TestHdfsClient, DriverTypes);
+TYPED_TEST_CASE(TestHadoopFileSystem, DriverTypes);
-TYPED_TEST(TestHdfsClient, ConnectsAgain) {
+TYPED_TEST(TestHadoopFileSystem, ConnectsAgain) {
SKIP_IF_NO_DRIVER();
- std::shared_ptr<HdfsClient> client;
- ASSERT_OK(HdfsClient::Connect(&this->conf_, &client));
+ std::shared_ptr<HadoopFileSystem> client;
+ ASSERT_OK(HadoopFileSystem::Connect(&this->conf_, &client));
ASSERT_OK(client->Disconnect());
}
-TYPED_TEST(TestHdfsClient, MakeDirectory) {
+TYPED_TEST(TestHadoopFileSystem, MakeDirectory) {
SKIP_IF_NO_DRIVER();
std::string path = this->ScratchPath("create-directory");
@@ -190,7 +190,7 @@ TYPED_TEST(TestHdfsClient, MakeDirectory) {
ASSERT_RAISES(IOError, this->client_->ListDirectory(path, &listing));
}
-TYPED_TEST(TestHdfsClient, GetCapacityUsed) {
+TYPED_TEST(TestHadoopFileSystem, GetCapacityUsed) {
SKIP_IF_NO_DRIVER();
// Who knows what is actually in your DFS cluster, but expect it to have
@@ -203,7 +203,7 @@ TYPED_TEST(TestHdfsClient, GetCapacityUsed) {
ASSERT_LT(0, nbytes);
}
-TYPED_TEST(TestHdfsClient, GetPathInfo) {
+TYPED_TEST(TestHadoopFileSystem, GetPathInfo) {
SKIP_IF_NO_DRIVER();
HdfsPathInfo info;
@@ -233,7 +233,7 @@ TYPED_TEST(TestHdfsClient, GetPathInfo) {
ASSERT_EQ(size, info.size);
}
-TYPED_TEST(TestHdfsClient, AppendToFile) {
+TYPED_TEST(TestHadoopFileSystem, AppendToFile) {
SKIP_IF_NO_DRIVER();
ASSERT_OK(this->MakeScratchDir());
@@ -252,7 +252,7 @@ TYPED_TEST(TestHdfsClient, AppendToFile) {
ASSERT_EQ(size * 2, info.size);
}
-TYPED_TEST(TestHdfsClient, ListDirectory) {
+TYPED_TEST(TestHadoopFileSystem, ListDirectory) {
SKIP_IF_NO_DRIVER();
const int size = 100;
@@ -292,7 +292,7 @@ TYPED_TEST(TestHdfsClient, ListDirectory) {
}
}
-TYPED_TEST(TestHdfsClient, ReadableMethods) {
+TYPED_TEST(TestHadoopFileSystem, ReadableMethods) {
SKIP_IF_NO_DRIVER();
ASSERT_OK(this->MakeScratchDir());
@@ -339,7 +339,7 @@ TYPED_TEST(TestHdfsClient, ReadableMethods) {
ASSERT_EQ(60, position);
}
-TYPED_TEST(TestHdfsClient, LargeFile) {
+TYPED_TEST(TestHadoopFileSystem, LargeFile) {
SKIP_IF_NO_DRIVER();
ASSERT_OK(this->MakeScratchDir());
@@ -374,7 +374,7 @@ TYPED_TEST(TestHdfsClient, LargeFile) {
ASSERT_EQ(size, bytes_read);
}
-TYPED_TEST(TestHdfsClient, RenameFile) {
+TYPED_TEST(TestHadoopFileSystem, RenameFile) {
SKIP_IF_NO_DRIVER();
ASSERT_OK(this->MakeScratchDir());
@@ -391,7 +391,32 @@ TYPED_TEST(TestHdfsClient, RenameFile) {
ASSERT_TRUE(this->client_->Exists(dst_path));
}
-TYPED_TEST(TestHdfsClient, ThreadSafety) {
+TYPED_TEST(TestHadoopFileSystem, ChmodChown) {
+ SKIP_IF_NO_DRIVER();
+ ASSERT_OK(this->MakeScratchDir());
+
+ auto path = this->ScratchPath("path-to-chmod");
+
+ int16_t mode = 0755;
+ const int size = 100;
+
+ std::vector<uint8_t> data = RandomData(size);
+ ASSERT_OK(this->WriteDummyFile(path, data.data(), size));
+
+ HdfsPathInfo info;
+ ASSERT_OK(this->client_->Chmod(path, mode));
+ ASSERT_OK(this->client_->GetPathInfo(path, &info));
+ ASSERT_EQ(mode, info.permissions);
+
+ std::string owner = "hadoop";
+ std::string group = "hadoop";
+ ASSERT_OK(this->client_->Chown(path, owner.c_str(), group.c_str()));
+ ASSERT_OK(this->client_->GetPathInfo(path, &info));
+ ASSERT_EQ("hadoop", info.owner);
+ ASSERT_EQ("hadoop", info.group);
+}
+
+TYPED_TEST(TestHadoopFileSystem, ThreadSafety) {
SKIP_IF_NO_DRIVER();
ASSERT_OK(this->MakeScratchDir());
http://git-wip-us.apache.org/repos/asf/arrow/blob/e1d574c7/python/doc/source/api.rst
----------------------------------------------------------------------
diff --git a/python/doc/source/api.rst b/python/doc/source/api.rst
index 6f26076..fd1cb72 100644
--- a/python/doc/source/api.rst
+++ b/python/doc/source/api.rst
@@ -164,15 +164,17 @@ Input / Output and Shared Memory
create_memory_map
PythonFile
-Filesystems
------------
+File Systems
+------------
.. autosummary::
:toctree: generated/
hdfs.connect
- HadoopFilesystem
- LocalFilesystem
+ LocalFileSystem
+
+.. class:: HadoopFileSystem
+ :noindex:
.. _api.ipc:
http://git-wip-us.apache.org/repos/asf/arrow/blob/e1d574c7/python/doc/source/conf.py
----------------------------------------------------------------------
diff --git a/python/doc/source/conf.py b/python/doc/source/conf.py
index d71b668..25e6d5e 100644
--- a/python/doc/source/conf.py
+++ b/python/doc/source/conf.py
@@ -62,7 +62,8 @@ extensions = [
]
# Show members for classes in .. autosummary
-autodoc_default_flags = ['members', 'undoc-members', 'show-inheritance', 'inherited-members']
+autodoc_default_flags = ['members', 'undoc-members', 'show-inheritance',
+ 'inherited-members']
# numpydoc configuration
napoleon_use_rtype = False
http://git-wip-us.apache.org/repos/asf/arrow/blob/e1d574c7/python/doc/source/filesystems.rst
----------------------------------------------------------------------
diff --git a/python/doc/source/filesystems.rst b/python/doc/source/filesystems.rst
index 78f6f2a..c0530f9 100644
--- a/python/doc/source/filesystems.rst
+++ b/python/doc/source/filesystems.rst
@@ -15,8 +15,8 @@
.. specific language governing permissions and limitations
.. under the License.
-Filesystem Interfaces
-=====================
+File System Interfaces
+======================
In this section, we discuss filesystem-like interfaces in PyArrow.
@@ -31,10 +31,11 @@ System. You connect like so:
.. code-block:: python
import pyarrow as pa
- hdfs = pa.hdfs.connect(host, port, user=user, kerb_ticket=ticket_cache_path)
- type(hdfs)
+ fs = pa.hdfs.connect(host, port, user=user, kerb_ticket=ticket_cache_path)
+ with fs.open(path, 'rb') as f:
+ # Do something with f
-By default, ``pyarrow.hdfs.HadoopFilesystem`` uses libhdfs, a JNI-based
+By default, ``pyarrow.hdfs.HadoopFileSystem`` uses libhdfs, a JNI-based
interface to the Java Hadoop client. This library is loaded **at runtime**
(rather than at link / library load time, since the library may not be in your
LD_LIBRARY_PATH), and relies on some environment variables.
@@ -57,5 +58,33 @@ You can also use libhdfs3, a thirdparty C++ library for HDFS from Pivotal Labs:
.. code-block:: python
- hdfs3 = pa.hdfs.connect(host, port, user=user, kerb_ticket=ticket_cache_path,
- driver='libhdfs3')
+ fs = pa.hdfs.connect(host, port, user=user, kerb_ticket=ticket_cache_path,
+ driver='libhdfs3')
+
+HDFS API
+~~~~~~~~
+
+.. currentmodule:: pyarrow
+
+.. autosummary::
+ :toctree: generated/
+
+ hdfs.connect
+ HadoopFileSystem.cat
+ HadoopFileSystem.chmod
+ HadoopFileSystem.chown
+ HadoopFileSystem.delete
+ HadoopFileSystem.df
+ HadoopFileSystem.disk_usage
+ HadoopFileSystem.download
+ HadoopFileSystem.exists
+ HadoopFileSystem.get_capacity
+ HadoopFileSystem.get_space_used
+ HadoopFileSystem.info
+ HadoopFileSystem.ls
+ HadoopFileSystem.mkdir
+ HadoopFileSystem.open
+ HadoopFileSystem.rename
+ HadoopFileSystem.rm
+ HadoopFileSystem.upload
+ HdfsFile
http://git-wip-us.apache.org/repos/asf/arrow/blob/e1d574c7/python/doc/source/memory.rst
----------------------------------------------------------------------
diff --git a/python/doc/source/memory.rst b/python/doc/source/memory.rst
index ccc6298..f189199 100644
--- a/python/doc/source/memory.rst
+++ b/python/doc/source/memory.rst
@@ -226,10 +226,3 @@ file interfaces that can read and write to Arrow Buffers.
reader.read(7)
These have similar semantics to Python's built-in ``io.BytesIO``.
-
-Hadoop Filesystem
------------------
-
-:class:`~pyarrow.HdfsFile` is an implementation of :class:`~pyarrow.NativeFile`
-that can read and write to the Hadoop filesytem. Read more in the
-:ref:`Filesystems Section <hdfs>`.
http://git-wip-us.apache.org/repos/asf/arrow/blob/e1d574c7/python/pyarrow/__init__.py
----------------------------------------------------------------------
diff --git a/python/pyarrow/__init__.py b/python/pyarrow/__init__.py
index 42e5803..8d4a214 100644
--- a/python/pyarrow/__init__.py
+++ b/python/pyarrow/__init__.py
@@ -88,9 +88,9 @@ from pyarrow.lib import (ArrowException,
ArrowTypeError)
-from pyarrow.filesystem import Filesystem, LocalFilesystem
+from pyarrow.filesystem import FileSystem, LocalFileSystem
-from pyarrow.hdfs import HadoopFilesystem
+from pyarrow.hdfs import HadoopFileSystem
import pyarrow.hdfs as hdfs
from pyarrow.ipc import (Message, MessageReader,
@@ -103,7 +103,7 @@ from pyarrow.ipc import (Message, MessageReader,
open_file,
serialize_pandas, deserialize_pandas)
-localfs = LocalFilesystem.get_instance()
+localfs = LocalFileSystem.get_instance()
# ----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/arrow/blob/e1d574c7/python/pyarrow/filesystem.py
----------------------------------------------------------------------
diff --git a/python/pyarrow/filesystem.py b/python/pyarrow/filesystem.py
index 4b8ca32..8d2d8fc 100644
--- a/python/pyarrow/filesystem.py
+++ b/python/pyarrow/filesystem.py
@@ -22,10 +22,21 @@ import posixpath
from pyarrow.util import implements
-class Filesystem(object):
+class FileSystem(object):
"""
Abstract filesystem interface
"""
+ def cat(self, path):
+ """
+ Return contents of file as a bytes object
+
+ Returns
+ -------
+ contents : bytes
+ """
+ with self.open(path, 'rb') as f:
+ return f.read()
+
def ls(self, path):
"""
Return list of file paths
@@ -44,12 +55,68 @@ class Filesystem(object):
"""
raise NotImplementedError
+ def disk_usage(self, path):
+ """
+ Compute bytes used by all contents under indicated path in file tree
+
+ Parameters
+ ----------
+ path : string
+ Can be a file path or directory
+
+ Returns
+ -------
+ usage : int
+ """
+ path_info = self.stat(path)
+ if path_info['kind'] == 'file':
+ return path_info['size']
+
+ total = 0
+ for root, directories, files in self.walk(path):
+ for child_path in files:
+ abspath = self._path_join(root, child_path)
+ total += self.stat(abspath)['size']
+
+ return total
+
+ def _path_join(self, *args):
+ return self.pathsep.join(args)
+
+ def stat(self, path):
+ """
+
+ Returns
+ -------
+ stat : dict
+ """
+ raise NotImplementedError('FileSystem.stat')
+
def rm(self, path, recursive=False):
"""
- Alias for Filesystem.delete
+ Alias for FileSystem.delete
"""
return self.delete(path, recursive=recursive)
+ def mv(self, path, new_path):
+ """
+ Alias for FileSystem.rename
+ """
+ return self.rename(path, new_path)
+
+ def rename(self, path, new_path):
+ """
+ Rename file, like UNIX mv command
+
+ Parameters
+ ----------
+ path : string
+ Path to alter
+ new_path : string
+ Path to move to
+ """
+ raise NotImplementedError('FileSystem.rename')
+
def mkdir(self, path, create_parents=True):
raise NotImplementedError
@@ -113,40 +180,40 @@ class Filesystem(object):
return '/'
-class LocalFilesystem(Filesystem):
+class LocalFileSystem(FileSystem):
_instance = None
@classmethod
def get_instance(cls):
if cls._instance is None:
- cls._instance = LocalFilesystem()
+ cls._instance = LocalFileSystem()
return cls._instance
- @implements(Filesystem.ls)
+ @implements(FileSystem.ls)
def ls(self, path):
return sorted(pjoin(path, x) for x in os.listdir(path))
- @implements(Filesystem.mkdir)
+ @implements(FileSystem.mkdir)
def mkdir(self, path, create_parents=True):
if create_parents:
os.makedirs(path)
else:
os.mkdir(path)
- @implements(Filesystem.isdir)
+ @implements(FileSystem.isdir)
def isdir(self, path):
return os.path.isdir(path)
- @implements(Filesystem.isfile)
+ @implements(FileSystem.isfile)
def isfile(self, path):
return os.path.isfile(path)
- @implements(Filesystem.exists)
+ @implements(FileSystem.exists)
def exists(self, path):
return os.path.exists(path)
- @implements(Filesystem.open)
+ @implements(FileSystem.open)
def open(self, path, mode='rb'):
"""
Open file for reading or writing
@@ -164,7 +231,7 @@ class LocalFilesystem(Filesystem):
return os.walk(top_dir)
-class DaskFilesystem(Filesystem):
+class DaskFileSystem(FileSystem):
"""
Wraps s3fs Dask filesystem implementation like s3fs, gcsfs, etc.
"""
@@ -172,23 +239,23 @@ class DaskFilesystem(Filesystem):
def __init__(self, fs):
self.fs = fs
- @implements(Filesystem.isdir)
+ @implements(FileSystem.isdir)
def isdir(self, path):
raise NotImplementedError("Unsupported file system API")
- @implements(Filesystem.isfile)
+ @implements(FileSystem.isfile)
def isfile(self, path):
raise NotImplementedError("Unsupported file system API")
- @implements(Filesystem.delete)
+ @implements(FileSystem.delete)
def delete(self, path, recursive=False):
return self.fs.rm(path, recursive=recursive)
- @implements(Filesystem.mkdir)
+ @implements(FileSystem.mkdir)
def mkdir(self, path):
return self.fs.mkdir(path)
- @implements(Filesystem.open)
+ @implements(FileSystem.open)
def open(self, path, mode='rb'):
"""
Open file for reading or writing
@@ -205,9 +272,9 @@ class DaskFilesystem(Filesystem):
return self.fs.walk(top_path)
-class S3FSWrapper(DaskFilesystem):
+class S3FSWrapper(DaskFileSystem):
- @implements(Filesystem.isdir)
+ @implements(FileSystem.isdir)
def isdir(self, path):
try:
contents = self.fs.ls(path)
@@ -218,7 +285,7 @@ class S3FSWrapper(DaskFilesystem):
except OSError:
return False
- @implements(Filesystem.isfile)
+ @implements(FileSystem.isfile)
def isfile(self, path):
try:
contents = self.fs.ls(path)
http://git-wip-us.apache.org/repos/asf/arrow/blob/e1d574c7/python/pyarrow/hdfs.py
----------------------------------------------------------------------
diff --git a/python/pyarrow/hdfs.py b/python/pyarrow/hdfs.py
index 3240f99..855cc1e 100644
--- a/python/pyarrow/hdfs.py
+++ b/python/pyarrow/hdfs.py
@@ -18,13 +18,13 @@
import posixpath
from pyarrow.util import implements
-from pyarrow.filesystem import Filesystem
+from pyarrow.filesystem import FileSystem
import pyarrow.lib as lib
-class HadoopFilesystem(lib._HdfsClient, Filesystem):
+class HadoopFileSystem(lib.HadoopFileSystem, FileSystem):
"""
- Filesystem interface for HDFS cluster. See pyarrow.hdfs.connect for full
+ FileSystem interface for HDFS cluster. See pyarrow.hdfs.connect for full
connection details
"""
@@ -32,21 +32,25 @@ class HadoopFilesystem(lib._HdfsClient, Filesystem):
driver='libhdfs'):
self._connect(host, port, user, kerb_ticket, driver)
- @implements(Filesystem.isdir)
+ @implements(FileSystem.isdir)
def isdir(self, path):
- return lib._HdfsClient.isdir(self, path)
+ return super(HadoopFileSystem, self).isdir(path)
- @implements(Filesystem.isfile)
+ @implements(FileSystem.isfile)
def isfile(self, path):
- return lib._HdfsClient.isfile(self, path)
+ return super(HadoopFileSystem, self).isfile(path)
- @implements(Filesystem.delete)
+ @implements(FileSystem.delete)
def delete(self, path, recursive=False):
- return lib._HdfsClient.delete(self, path, recursive)
+ return super(HadoopFileSystem, self).delete(path, recursive)
- @implements(Filesystem.mkdir)
+ @implements(FileSystem.mkdir)
def mkdir(self, path, create_parents=True):
- return lib._HdfsClient.mkdir(self, path)
+ return super(HadoopFileSystem, self).mkdir(path)
+
+ @implements(FileSystem.rename)
+ def rename(self, path, new_path):
+ return super(HadoopFileSystem, self).rename(path, new_path)
def ls(self, path, detail=False):
"""
@@ -62,7 +66,7 @@ class HadoopFilesystem(lib._HdfsClient, Filesystem):
-------
result : list of dicts (detail=True) or strings (detail=False)
"""
- return lib._HdfsClient.ls(self, path, detail)
+ return super(HadoopFileSystem, self).ls(path, detail)
def walk(self, top_path):
"""
@@ -82,7 +86,7 @@ class HadoopFilesystem(lib._HdfsClient, Filesystem):
directories, files = _libhdfs_walk_files_dirs(top_path, contents)
yield top_path, directories, files
for dirname in directories:
- for tup in self.walk(dirname):
+ for tup in self.walk(self._path_join(top_path, dirname)):
yield tup
@@ -126,8 +130,8 @@ def connect(host="default", port=0, user=None, kerb_ticket=None,
Returns
-------
- filesystem : HadoopFilesystem
+ filesystem : HadoopFileSystem
"""
- fs = HadoopFilesystem(host=host, port=port, user=user,
+ fs = HadoopFileSystem(host=host, port=port, user=user,
kerb_ticket=kerb_ticket, driver=driver)
return fs
http://git-wip-us.apache.org/repos/asf/arrow/blob/e1d574c7/python/pyarrow/includes/libarrow.pxd
----------------------------------------------------------------------
diff --git a/python/pyarrow/includes/libarrow.pxd b/python/pyarrow/includes/libarrow.pxd
index db6770f..8d7e279 100644
--- a/python/pyarrow/includes/libarrow.pxd
+++ b/python/pyarrow/includes/libarrow.pxd
@@ -413,6 +413,10 @@ cdef extern from "arrow/io/interfaces.h" namespace "arrow::io" nogil:
ObjectType_FILE" arrow::io::ObjectType::FILE"
ObjectType_DIRECTORY" arrow::io::ObjectType::DIRECTORY"
+ cdef cppclass FileStatistics:
+ int64_t size
+ ObjectType kind
+
cdef cppclass FileInterface:
CStatus Close()
CStatus Tell(int64_t* position)
@@ -450,6 +454,9 @@ cdef extern from "arrow/io/interfaces.h" namespace "arrow::io" nogil:
WriteableFile):
pass
+ cdef cppclass FileSystem:
+ CStatus Stat(const c_string& path, FileStatistics* stat)
+
cdef extern from "arrow/io/file.h" namespace "arrow::io" nogil:
@@ -517,10 +524,10 @@ cdef extern from "arrow/io/hdfs.h" namespace "arrow::io" nogil:
cdef cppclass HdfsOutputStream(OutputStream):
pass
- cdef cppclass CHdfsClient" arrow::io::HdfsClient":
+ cdef cppclass CHadoopFileSystem" arrow::io::HadoopFileSystem"(FileSystem):
@staticmethod
CStatus Connect(const HdfsConnectionConfig* config,
- shared_ptr[CHdfsClient]* client)
+ shared_ptr[CHadoopFileSystem]* client)
CStatus MakeDirectory(const c_string& path)
@@ -530,6 +537,10 @@ cdef extern from "arrow/io/hdfs.h" namespace "arrow::io" nogil:
c_bool Exists(const c_string& path)
+ CStatus Chmod(const c_string& path, int mode)
+ CStatus Chown(const c_string& path, const char* owner,
+ const char* group)
+
CStatus GetCapacity(int64_t* nbytes)
CStatus GetUsed(int64_t* nbytes)
http://git-wip-us.apache.org/repos/asf/arrow/blob/e1d574c7/python/pyarrow/io-hdfs.pxi
----------------------------------------------------------------------
diff --git a/python/pyarrow/io-hdfs.pxi b/python/pyarrow/io-hdfs.pxi
new file mode 100644
index 0000000..8ac4e8c
--- /dev/null
+++ b/python/pyarrow/io-hdfs.pxi
@@ -0,0 +1,468 @@
+# 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.
+
+# ----------------------------------------------------------------------
+# HDFS IO implementation
+
+_HDFS_PATH_RE = re.compile('hdfs://(.*):(\d+)(.*)')
+
+try:
+ # Python 3
+ from queue import Queue, Empty as QueueEmpty, Full as QueueFull
+except ImportError:
+ from Queue import Queue, Empty as QueueEmpty, Full as QueueFull
+
+
+def have_libhdfs():
+ try:
+ check_status(HaveLibHdfs())
+ return True
+ except:
+ return False
+
+
+def have_libhdfs3():
+ try:
+ check_status(HaveLibHdfs3())
+ return True
+ except:
+ return False
+
+
+def strip_hdfs_abspath(path):
+ m = _HDFS_PATH_RE.match(path)
+ if m:
+ return m.group(3)
+ else:
+ return path
+
+
+cdef class HadoopFileSystem:
+ cdef:
+ shared_ptr[CHadoopFileSystem] client
+
+ cdef readonly:
+ bint is_open
+
+ def __cinit__(self):
+ pass
+
+ def _connect(self, host, port, user, kerb_ticket, driver):
+ cdef HdfsConnectionConfig conf
+
+ if host is not None:
+ conf.host = tobytes(host)
+ conf.port = port
+ if user is not None:
+ conf.user = tobytes(user)
+ if kerb_ticket is not None:
+ conf.kerb_ticket = tobytes(kerb_ticket)
+
+ if driver == 'libhdfs':
+ check_status(HaveLibHdfs())
+ conf.driver = HdfsDriver_LIBHDFS
+ else:
+ check_status(HaveLibHdfs3())
+ conf.driver = HdfsDriver_LIBHDFS3
+
+ with nogil:
+ check_status(CHadoopFileSystem.Connect(&conf, &self.client))
+ self.is_open = True
+
+ @classmethod
+ def connect(cls, *args, **kwargs):
+ return cls(*args, **kwargs)
+
+ def __dealloc__(self):
+ if self.is_open:
+ self.close()
+
+ def close(self):
+ """
+ Disconnect from the HDFS cluster
+ """
+ self._ensure_client()
+ with nogil:
+ check_status(self.client.get().Disconnect())
+ self.is_open = False
+
+ cdef _ensure_client(self):
+ if self.client.get() == NULL:
+ raise IOError('HDFS client improperly initialized')
+ elif not self.is_open:
+ raise IOError('HDFS client is closed')
+
+ def exists(self, path):
+ """
+ Returns True if the path is known to the cluster, False if it does not
+ (or there is an RPC error)
+ """
+ self._ensure_client()
+
+ cdef c_string c_path = tobytes(path)
+ cdef c_bool result
+ with nogil:
+ result = self.client.get().Exists(c_path)
+ return result
+
+ def isdir(self, path):
+ cdef HdfsPathInfo info
+ self._path_info(path, &info)
+ return info.kind == ObjectType_DIRECTORY
+
+ def isfile(self, path):
+ cdef HdfsPathInfo info
+ self._path_info(path, &info)
+ return info.kind == ObjectType_FILE
+
+ def get_capacity(self):
+ """
+ Get reported total capacity of file system
+
+ Returns
+ -------
+ capacity : int
+ """
+ cdef int64_t capacity = 0
+ with nogil:
+ check_status(self.client.get().GetCapacity(&capacity))
+ return capacity
+
+ def get_space_used(self):
+ """
+ Get space used on file system
+
+ Returns
+ -------
+ space_used : int
+ """
+ cdef int64_t space_used = 0
+ with nogil:
+ check_status(self.client.get().GetUsed(&space_used))
+ return space_used
+
+ def df(self):
+ """
+ Return free space on disk, like the UNIX df command
+
+ Returns
+ -------
+ space : int
+ """
+ return self.get_capacity() - self.get_space_used()
+
+ def rename(self, path, new_path):
+ cdef c_string c_path = tobytes(path)
+ cdef c_string c_new_path = tobytes(new_path)
+ with nogil:
+ check_status(self.client.get().Rename(c_path, c_new_path))
+
+ def info(self, path):
+ """
+ Return detailed HDFS information for path
+
+ Parameters
+ ----------
+ path : string
+ Path to file or directory
+
+ Returns
+ -------
+ path_info : dict
+ """
+ cdef HdfsPathInfo info
+ self._path_info(path, &info)
+ return {
+ 'path': frombytes(info.name),
+ 'owner': frombytes(info.owner),
+ 'group': frombytes(info.group),
+ 'size': info.size,
+ 'block_size': info.block_size,
+ 'last_modified': info.last_modified_time,
+ 'last_accessed': info.last_access_time,
+ 'replication': info.replication,
+ 'permissions': info.permissions,
+ 'kind': ('directory' if info.kind == ObjectType_DIRECTORY
+ else 'file')
+ }
+
+ def stat(self, path):
+ """
+ Return basic file system statistics about path
+
+ Parameters
+ ----------
+ path : string
+ Path to file or directory
+
+ Returns
+ -------
+ stat : dict
+ """
+ cdef FileStatistics info
+ cdef c_string c_path = tobytes(path)
+ with nogil:
+ check_status(self.client.get()
+ .Stat(c_path, &info))
+ return {
+ 'size': info.size,
+ 'kind': ('directory' if info.kind == ObjectType_DIRECTORY
+ else 'file')
+ }
+
+ cdef _path_info(self, path, HdfsPathInfo* info):
+ cdef c_string c_path = tobytes(path)
+
+ with nogil:
+ check_status(self.client.get()
+ .GetPathInfo(c_path, info))
+
+
+ def ls(self, path, bint full_info):
+ cdef:
+ c_string c_path = tobytes(path)
+ vector[HdfsPathInfo] listing
+ list results = []
+ int i
+
+ self._ensure_client()
+
+ with nogil:
+ check_status(self.client.get()
+ .ListDirectory(c_path, &listing))
+
+ cdef const HdfsPathInfo* info
+ for i in range(<int> listing.size()):
+ info = &listing[i]
+
+ # Try to trim off the hdfs://HOST:PORT piece
+ name = strip_hdfs_abspath(frombytes(info.name))
+
+ if full_info:
+ kind = ('file' if info.kind == ObjectType_FILE
+ else 'directory')
+
+ results.append({
+ 'kind': kind,
+ 'name': name,
+ 'owner': frombytes(info.owner),
+ 'group': frombytes(info.group),
+ 'list_modified_time': info.last_modified_time,
+ 'list_access_time': info.last_access_time,
+ 'size': info.size,
+ 'replication': info.replication,
+ 'block_size': info.block_size,
+ 'permissions': info.permissions
+ })
+ else:
+ results.append(name)
+
+ return results
+
+ def chmod(self, path, mode):
+ """
+ Change file permissions
+
+ Parameters
+ ----------
+ path : string
+ absolute path to file or directory
+ mode : int
+ POSIX-like bitmask
+ """
+ self._ensure_client()
+ cdef c_string c_path = tobytes(path)
+ cdef int c_mode = mode
+ with nogil:
+ check_status(self.client.get()
+ .Chmod(c_path, c_mode))
+
+ def chown(self, path, owner=None, group=None):
+ """
+ Change file permissions
+
+ Parameters
+ ----------
+ path : string
+ absolute path to file or directory
+ owner : string, default None
+ New owner, None for no change
+ group : string, default None
+ New group, None for no change
+ """
+ cdef:
+ c_string c_path
+ c_string c_owner
+ c_string c_group
+ const char* c_owner_ptr = NULL
+ const char* c_group_ptr = NULL
+
+ self._ensure_client()
+
+ c_path = tobytes(path)
+ if owner is not None:
+ c_owner = tobytes(owner)
+ c_owner_ptr = c_owner.c_str()
+
+ if group is not None:
+ c_group = tobytes(group)
+ c_group_ptr = c_group.c_str()
+
+ with nogil:
+ check_status(self.client.get()
+ .Chown(c_path, c_owner_ptr, c_group_ptr))
+
+ def mkdir(self, path):
+ """
+ Create indicated directory and any necessary parent directories
+ """
+ self._ensure_client()
+ cdef c_string c_path = tobytes(path)
+ with nogil:
+ check_status(self.client.get()
+ .MakeDirectory(c_path))
+
+ def delete(self, path, bint recursive=False):
+ """
+ Delete the indicated file or directory
+
+ Parameters
+ ----------
+ path : string
+ recursive : boolean, default False
+ If True, also delete child paths for directories
+ """
+ self._ensure_client()
+
+ cdef c_string c_path = tobytes(path)
+ with nogil:
+ check_status(self.client.get()
+ .Delete(c_path, recursive == 1))
+
+ def open(self, path, mode='rb', buffer_size=None, replication=None,
+ default_block_size=None):
+ """
+ Open HDFS file for reading or writing
+
+ Parameters
+ ----------
+ mode : string
+ Must be one of 'rb', 'wb', 'ab'
+
+ Returns
+ -------
+ handle : HdfsFile
+ """
+ self._ensure_client()
+
+ cdef HdfsFile out = HdfsFile()
+
+ if mode not in ('rb', 'wb', 'ab'):
+ raise Exception("Mode must be 'rb' (read), "
+ "'wb' (write, new file), or 'ab' (append)")
+
+ cdef c_string c_path = tobytes(path)
+ cdef c_bool append = False
+
+ # 0 in libhdfs means "use the default"
+ cdef int32_t c_buffer_size = buffer_size or 0
+ cdef int16_t c_replication = replication or 0
+ cdef int64_t c_default_block_size = default_block_size or 0
+
+ cdef shared_ptr[HdfsOutputStream] wr_handle
+ cdef shared_ptr[HdfsReadableFile] rd_handle
+
+ if mode in ('wb', 'ab'):
+ if mode == 'ab':
+ append = True
+
+ with nogil:
+ check_status(
+ self.client.get()
+ .OpenWriteable(c_path, append, c_buffer_size,
+ c_replication, c_default_block_size,
+ &wr_handle))
+
+ out.wr_file = <shared_ptr[OutputStream]> wr_handle
+
+ out.is_readable = False
+ out.is_writeable = 1
+ else:
+ with nogil:
+ check_status(self.client.get()
+ .OpenReadable(c_path, &rd_handle))
+
+ out.rd_file = <shared_ptr[RandomAccessFile]> rd_handle
+ out.is_readable = True
+ out.is_writeable = 0
+
+ if c_buffer_size == 0:
+ c_buffer_size = 2 ** 16
+
+ out.mode = mode
+ out.buffer_size = c_buffer_size
+ out.parent = _HdfsFileNanny(self, out)
+ out.is_open = True
+ out.own_file = True
+
+ return out
+
+ def download(self, path, stream, buffer_size=None):
+ with self.open(path, 'rb') as f:
+ f.download(stream, buffer_size=buffer_size)
+
+ def upload(self, path, stream, buffer_size=None):
+ """
+ Upload file-like object to HDFS path
+ """
+ with self.open(path, 'wb') as f:
+ f.upload(stream, buffer_size=buffer_size)
+
+
+# ARROW-404: Helper class to ensure that files are closed before the
+# client. During deallocation of the extension class, the attributes are
+# decref'd which can cause the client to get closed first if the file has the
+# last remaining reference
+cdef class _HdfsFileNanny:
+ cdef:
+ object client
+ object file_handle_ref
+
+ def __cinit__(self, client, file_handle):
+ import weakref
+ self.client = client
+ self.file_handle_ref = weakref.ref(file_handle)
+
+ def __dealloc__(self):
+ fh = self.file_handle_ref()
+ if fh:
+ fh.close()
+ # avoid cyclic GC
+ self.file_handle_ref = None
+ self.client = None
+
+
+cdef class HdfsFile(NativeFile):
+ cdef readonly:
+ int32_t buffer_size
+ object mode
+ object parent
+
+ cdef object __weakref__
+
+ def __dealloc__(self):
+ self.parent = None
http://git-wip-us.apache.org/repos/asf/arrow/blob/e1d574c7/python/pyarrow/io.pxi
----------------------------------------------------------------------
diff --git a/python/pyarrow/io.pxi b/python/pyarrow/io.pxi
index 01c987d..211c2a3 100644
--- a/python/pyarrow/io.pxi
+++ b/python/pyarrow/io.pxi
@@ -733,301 +733,3 @@ cdef get_writer(object source, shared_ptr[OutputStream]* writer):
else:
raise TypeError('Unable to read from object of type: {0}'
.format(type(source)))
-
-# ----------------------------------------------------------------------
-# HDFS IO implementation
-
-_HDFS_PATH_RE = re.compile('hdfs://(.*):(\d+)(.*)')
-
-try:
- # Python 3
- from queue import Queue, Empty as QueueEmpty, Full as QueueFull
-except ImportError:
- from Queue import Queue, Empty as QueueEmpty, Full as QueueFull
-
-
-def have_libhdfs():
- try:
- check_status(HaveLibHdfs())
- return True
- except:
- return False
-
-
-def have_libhdfs3():
- try:
- check_status(HaveLibHdfs3())
- return True
- except:
- return False
-
-
-def strip_hdfs_abspath(path):
- m = _HDFS_PATH_RE.match(path)
- if m:
- return m.group(3)
- else:
- return path
-
-
-cdef class _HdfsClient:
- cdef:
- shared_ptr[CHdfsClient] client
-
- cdef readonly:
- bint is_open
-
- def __cinit__(self):
- pass
-
- def _connect(self, host, port, user, kerb_ticket, driver):
- cdef HdfsConnectionConfig conf
-
- if host is not None:
- conf.host = tobytes(host)
- conf.port = port
- if user is not None:
- conf.user = tobytes(user)
- if kerb_ticket is not None:
- conf.kerb_ticket = tobytes(kerb_ticket)
-
- if driver == 'libhdfs':
- check_status(HaveLibHdfs())
- conf.driver = HdfsDriver_LIBHDFS
- else:
- check_status(HaveLibHdfs3())
- conf.driver = HdfsDriver_LIBHDFS3
-
- with nogil:
- check_status(CHdfsClient.Connect(&conf, &self.client))
- self.is_open = True
-
- @classmethod
- def connect(cls, *args, **kwargs):
- return cls(*args, **kwargs)
-
- def __dealloc__(self):
- if self.is_open:
- self.close()
-
- def close(self):
- """
- Disconnect from the HDFS cluster
- """
- self._ensure_client()
- with nogil:
- check_status(self.client.get().Disconnect())
- self.is_open = False
-
- cdef _ensure_client(self):
- if self.client.get() == NULL:
- raise IOError('HDFS client improperly initialized')
- elif not self.is_open:
- raise IOError('HDFS client is closed')
-
- def exists(self, path):
- """
- Returns True if the path is known to the cluster, False if it does not
- (or there is an RPC error)
- """
- self._ensure_client()
-
- cdef c_string c_path = tobytes(path)
- cdef c_bool result
- with nogil:
- result = self.client.get().Exists(c_path)
- return result
-
- def isdir(self, path):
- cdef HdfsPathInfo info
- self._path_info(path, &info)
- return info.kind == ObjectType_DIRECTORY
-
- def isfile(self, path):
- cdef HdfsPathInfo info
- self._path_info(path, &info)
- return info.kind == ObjectType_FILE
-
- cdef _path_info(self, path, HdfsPathInfo* info):
- cdef c_string c_path = tobytes(path)
-
- with nogil:
- check_status(self.client.get()
- .GetPathInfo(c_path, info))
-
-
- def ls(self, path, bint full_info):
- cdef:
- c_string c_path = tobytes(path)
- vector[HdfsPathInfo] listing
- list results = []
- int i
-
- self._ensure_client()
-
- with nogil:
- check_status(self.client.get()
- .ListDirectory(c_path, &listing))
-
- cdef const HdfsPathInfo* info
- for i in range(<int> listing.size()):
- info = &listing[i]
-
- # Try to trim off the hdfs://HOST:PORT piece
- name = strip_hdfs_abspath(frombytes(info.name))
-
- if full_info:
- kind = ('file' if info.kind == ObjectType_FILE
- else 'directory')
-
- results.append({
- 'kind': kind,
- 'name': name,
- 'owner': frombytes(info.owner),
- 'group': frombytes(info.group),
- 'list_modified_time': info.last_modified_time,
- 'list_access_time': info.last_access_time,
- 'size': info.size,
- 'replication': info.replication,
- 'block_size': info.block_size,
- 'permissions': info.permissions
- })
- else:
- results.append(name)
-
- return results
-
- def mkdir(self, path):
- """
- Create indicated directory and any necessary parent directories
- """
- self._ensure_client()
-
- cdef c_string c_path = tobytes(path)
- with nogil:
- check_status(self.client.get()
- .MakeDirectory(c_path))
-
- def delete(self, path, bint recursive=False):
- """
- Delete the indicated file or directory
-
- Parameters
- ----------
- path : string
- recursive : boolean, default False
- If True, also delete child paths for directories
- """
- self._ensure_client()
-
- cdef c_string c_path = tobytes(path)
- with nogil:
- check_status(self.client.get()
- .Delete(c_path, recursive == 1))
-
- def open(self, path, mode='rb', buffer_size=None, replication=None,
- default_block_size=None):
- """
- Parameters
- ----------
- mode : string, 'rb', 'wb', 'ab'
- """
- self._ensure_client()
-
- cdef HdfsFile out = HdfsFile()
-
- if mode not in ('rb', 'wb', 'ab'):
- raise Exception("Mode must be 'rb' (read), "
- "'wb' (write, new file), or 'ab' (append)")
-
- cdef c_string c_path = tobytes(path)
- cdef c_bool append = False
-
- # 0 in libhdfs means "use the default"
- cdef int32_t c_buffer_size = buffer_size or 0
- cdef int16_t c_replication = replication or 0
- cdef int64_t c_default_block_size = default_block_size or 0
-
- cdef shared_ptr[HdfsOutputStream] wr_handle
- cdef shared_ptr[HdfsReadableFile] rd_handle
-
- if mode in ('wb', 'ab'):
- if mode == 'ab':
- append = True
-
- with nogil:
- check_status(
- self.client.get()
- .OpenWriteable(c_path, append, c_buffer_size,
- c_replication, c_default_block_size,
- &wr_handle))
-
- out.wr_file = <shared_ptr[OutputStream]> wr_handle
-
- out.is_readable = False
- out.is_writeable = 1
- else:
- with nogil:
- check_status(self.client.get()
- .OpenReadable(c_path, &rd_handle))
-
- out.rd_file = <shared_ptr[RandomAccessFile]> rd_handle
- out.is_readable = True
- out.is_writeable = 0
-
- if c_buffer_size == 0:
- c_buffer_size = 2 ** 16
-
- out.mode = mode
- out.buffer_size = c_buffer_size
- out.parent = _HdfsFileNanny(self, out)
- out.is_open = True
- out.own_file = True
-
- return out
-
- def download(self, path, stream, buffer_size=None):
- with self.open(path, 'rb') as f:
- f.download(stream, buffer_size=buffer_size)
-
- def upload(self, path, stream, buffer_size=None):
- """
- Upload file-like object to HDFS path
- """
- with self.open(path, 'wb') as f:
- f.upload(stream, buffer_size=buffer_size)
-
-
-# ARROW-404: Helper class to ensure that files are closed before the
-# client. During deallocation of the extension class, the attributes are
-# decref'd which can cause the client to get closed first if the file has the
-# last remaining reference
-cdef class _HdfsFileNanny:
- cdef:
- object client
- object file_handle_ref
-
- def __cinit__(self, client, file_handle):
- import weakref
- self.client = client
- self.file_handle_ref = weakref.ref(file_handle)
-
- def __dealloc__(self):
- fh = self.file_handle_ref()
- if fh:
- fh.close()
- # avoid cyclic GC
- self.file_handle_ref = None
- self.client = None
-
-
-cdef class HdfsFile(NativeFile):
- cdef readonly:
- int32_t buffer_size
- object mode
- object parent
-
- cdef object __weakref__
-
- def __dealloc__(self):
- self.parent = None
http://git-wip-us.apache.org/repos/asf/arrow/blob/e1d574c7/python/pyarrow/lib.pyx
----------------------------------------------------------------------
diff --git a/python/pyarrow/lib.pyx b/python/pyarrow/lib.pyx
index 5990308..4df2fcd 100644
--- a/python/pyarrow/lib.pyx
+++ b/python/pyarrow/lib.pyx
@@ -114,6 +114,7 @@ include "table.pxi"
# File IO
include "io.pxi"
+include "io-hdfs.pxi"
# IPC / Messaging
include "ipc.pxi"
http://git-wip-us.apache.org/repos/asf/arrow/blob/e1d574c7/python/pyarrow/parquet.py
----------------------------------------------------------------------
diff --git a/python/pyarrow/parquet.py b/python/pyarrow/parquet.py
index c870412..6d39a23 100644
--- a/python/pyarrow/parquet.py
+++ b/python/pyarrow/parquet.py
@@ -22,7 +22,7 @@ import six
import numpy as np
-from pyarrow.filesystem import Filesystem, LocalFilesystem
+from pyarrow.filesystem import FileSystem, LocalFileSystem
from pyarrow._parquet import (ParquetReader, FileMetaData, # noqa
RowGroupMetaData, ParquetSchema,
ParquetWriter)
@@ -403,7 +403,7 @@ class ParquetManifest(object):
"""
def __init__(self, dirpath, filesystem=None, pathsep='/',
partition_scheme='hive'):
- self.filesystem = filesystem or LocalFilesystem.get_instance()
+ self.filesystem = filesystem or LocalFileSystem.get_instance()
self.pathsep = pathsep
self.dirpath = dirpath
self.partition_scheme = partition_scheme
@@ -506,7 +506,7 @@ class ParquetDataset(object):
----------
path_or_paths : str or List[str]
A directory name, single file name, or list of file names
- filesystem : Filesystem, default None
+ filesystem : FileSystem, default None
If nothing passed, paths assumed to be found in the local on-disk
filesystem
metadata : pyarrow.parquet.FileMetaData
@@ -522,7 +522,7 @@ class ParquetDataset(object):
def __init__(self, path_or_paths, filesystem=None, schema=None,
metadata=None, split_row_groups=False, validate_schema=True):
if filesystem is None:
- self.fs = LocalFilesystem.get_instance()
+ self.fs = LocalFileSystem.get_instance()
else:
self.fs = _ensure_filesystem(filesystem)
@@ -631,7 +631,7 @@ class ParquetDataset(object):
return keyvalues.get(b'pandas', None)
def _get_open_file_func(self):
- if self.fs is None or isinstance(self.fs, LocalFilesystem):
+ if self.fs is None or isinstance(self.fs, LocalFileSystem):
def open_file(path, meta=None):
return ParquetFile(path, metadata=meta,
common_metadata=self.common_metadata)
@@ -644,7 +644,7 @@ class ParquetDataset(object):
def _ensure_filesystem(fs):
- if not isinstance(fs, Filesystem):
+ if not isinstance(fs, FileSystem):
if type(fs).__name__ == 'S3FileSystem':
from pyarrow.filesystem import S3FSWrapper
return S3FSWrapper(fs)
@@ -716,7 +716,7 @@ def read_table(source, columns=None, nthreads=1, metadata=None,
Content of the file as a table (of columns)
"""
if is_string(source):
- fs = LocalFilesystem.get_instance()
+ fs = LocalFileSystem.get_instance()
if fs.isdir(source):
return fs.read_parquet(source, columns=columns,
metadata=metadata)
http://git-wip-us.apache.org/repos/asf/arrow/blob/e1d574c7/python/pyarrow/tests/test_hdfs.py
----------------------------------------------------------------------
diff --git a/python/pyarrow/tests/test_hdfs.py b/python/pyarrow/tests/test_hdfs.py
index 1026408..79638f2 100644
--- a/python/pyarrow/tests/test_hdfs.py
+++ b/python/pyarrow/tests/test_hdfs.py
@@ -72,7 +72,26 @@ class HdfsTestCases(object):
cls.hdfs.delete(cls.tmp_path, recursive=True)
cls.hdfs.close()
- def test_hdfs_close(self):
+ def test_cat(self):
+ path = pjoin(self.tmp_path, 'cat-test')
+
+ data = b'foobarbaz'
+ with self.hdfs.open(path, 'wb') as f:
+ f.write(data)
+
+ contents = self.hdfs.cat(path)
+ assert contents == data
+
+ def test_capacity_space(self):
+ capacity = self.hdfs.get_capacity()
+ space_used = self.hdfs.get_space_used()
+ disk_free = self.hdfs.df()
+
+ assert capacity > 0
+ assert capacity > space_used
+ assert disk_free == (capacity - space_used)
+
+ def test_close(self):
client = hdfs_test_client()
assert client.is_open
client.close()
@@ -81,7 +100,7 @@ class HdfsTestCases(object):
with pytest.raises(Exception):
client.ls('/')
- def test_hdfs_mkdir(self):
+ def test_mkdir(self):
path = pjoin(self.tmp_path, 'test-dir/test-dir')
parent_path = pjoin(self.tmp_path, 'test-dir')
@@ -91,7 +110,64 @@ class HdfsTestCases(object):
self.hdfs.delete(parent_path, recursive=True)
assert not self.hdfs.exists(path)
- def test_hdfs_ls(self):
+ def test_mv_rename(self):
+ path = pjoin(self.tmp_path, 'mv-test')
+ new_path = pjoin(self.tmp_path, 'mv-new-test')
+
+ data = b'foobarbaz'
+ with self.hdfs.open(path, 'wb') as f:
+ f.write(data)
+
+ assert self.hdfs.exists(path)
+ self.hdfs.mv(path, new_path)
+ assert not self.hdfs.exists(path)
+ assert self.hdfs.exists(new_path)
+
+ assert self.hdfs.cat(new_path) == data
+
+ self.hdfs.rename(new_path, path)
+ assert self.hdfs.cat(path) == data
+
+ def test_info(self):
+ path = pjoin(self.tmp_path, 'info-base')
+ file_path = pjoin(path, 'ex')
+ self.hdfs.mkdir(path)
+
+ data = b'foobarbaz'
+ with self.hdfs.open(file_path, 'wb') as f:
+ f.write(data)
+
+ path_info = self.hdfs.info(path)
+ file_path_info = self.hdfs.info(file_path)
+
+ assert path_info['kind'] == 'directory'
+
+ assert file_path_info['kind'] == 'file'
+ assert file_path_info['size'] == len(data)
+
+ def test_disk_usage(self):
+ path = pjoin(self.tmp_path, 'disk-usage-base')
+ p1 = pjoin(path, 'p1')
+ p2 = pjoin(path, 'p2')
+
+ subdir = pjoin(path, 'subdir')
+ p3 = pjoin(subdir, 'p3')
+
+ if self.hdfs.exists(path):
+ self.hdfs.delete(path, True)
+
+ self.hdfs.mkdir(path)
+ self.hdfs.mkdir(subdir)
+
+ data = b'foobarbaz'
+
+ for file_path in [p1, p2, p3]:
+ with self.hdfs.open(file_path, 'wb') as f:
+ f.write(data)
+
+ assert self.hdfs.disk_usage(path) == len(data) * 3
+
+ def test_ls(self):
base_path = pjoin(self.tmp_path, 'ls-test')
self.hdfs.mkdir(base_path)
@@ -106,7 +182,12 @@ class HdfsTestCases(object):
contents = sorted(self.hdfs.ls(base_path, False))
assert contents == [dir_path, f1_path]
- def test_hdfs_download_upload(self):
+ def test_chmod_chown(self):
+ path = pjoin(self.tmp_path, 'chmod-test')
+ with self.hdfs.open(path, 'wb') as f:
+ f.write(b'a' * 10)
+
+ def test_download_upload(self):
base_path = pjoin(self.tmp_path, 'upload-test')
data = b'foobarbaz'
@@ -120,7 +201,7 @@ class HdfsTestCases(object):
out_buf.seek(0)
assert out_buf.getvalue() == data
- def test_hdfs_file_context_manager(self):
+ def test_file_context_manager(self):
path = pjoin(self.tmp_path, 'ctx-manager')
data = b'foo'
@@ -132,7 +213,7 @@ class HdfsTestCases(object):
result = f.read(10)
assert result == data
- def test_hdfs_read_whole_file(self):
+ def test_read_whole_file(self):
path = pjoin(self.tmp_path, 'read-whole-file')
data = b'foo' * 1000
@@ -145,7 +226,7 @@ class HdfsTestCases(object):
assert result == data
@test_parquet.parquet
- def test_hdfs_read_multiple_parquet_files(self):
+ def test_read_multiple_parquet_files(self):
import pyarrow.parquet as pq
nfiles = 10
@@ -191,7 +272,7 @@ class TestLibHdfs(HdfsTestCases, unittest.TestCase):
if not pa.have_libhdfs():
pytest.fail('No libhdfs available on system')
- def test_hdfs_orphaned_file(self):
+ def test_orphaned_file(self):
hdfs = hdfs_test_client()
file_path = self._make_test_file(hdfs, 'orphaned_file_test', 'fname',
'foobarbaz')
http://git-wip-us.apache.org/repos/asf/arrow/blob/e1d574c7/python/pyarrow/tests/test_parquet.py
----------------------------------------------------------------------
diff --git a/python/pyarrow/tests/test_parquet.py b/python/pyarrow/tests/test_parquet.py
index 06265ca..ab3b26c 100644
--- a/python/pyarrow/tests/test_parquet.py
+++ b/python/pyarrow/tests/test_parquet.py
@@ -23,7 +23,7 @@ import json
import pytest
from pyarrow.compat import guid, u
-from pyarrow.filesystem import LocalFilesystem
+from pyarrow.filesystem import LocalFileSystem
import pyarrow as pa
from .pandas_examples import dataframe_with_arrays, dataframe_with_lists
@@ -700,7 +700,7 @@ def test_partition_set_dictionary_type():
@parquet
def test_read_partitioned_directory(tmpdir):
- fs = LocalFilesystem.get_instance()
+ fs = LocalFileSystem.get_instance()
base_path = str(tmpdir)
_partition_test_for_filesystem(fs, base_path)