mirrors/arrow
0
0 Fork 0
mirror of https://github.com/apache/arrow.git synced 2026-10-29 00:56:31 +08:00
Code Activity

Refactor HdfsClient code to support both libhdfs and libhdfs3 at runtime. Add driver option to Python interface

Browse Source
This commit is contained in:
Wes McKinney committed 2016-12-19 16:59:18 -05:00
1 parent c369709c4f
commit 7ae197ab07
11 files changed
+1139 -832
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

cpp/src/arrow/io/CMakeLists.txt
+1 -1
View File
@@ -75,7 +75,7 @@ if(ARROW_HDFS)
set(ARROW_HDFS_SRCS
hdfs.cc
libhdfs_shim.cc)
hdfs-internal.cc)
set_property(SOURCE ${ARROW_HDFS_SRCS}
APPEND_STRING PROPERTY
cpp/src/arrow/io/hdfs-internal.cc
+590
View File
@@ -0,0 +1,590 @@
// 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.
// This shim interface to libhdfs (for runtime shared library loading) has been
// adapted from the SFrame project, released under the ASF-compatible 3-clause
// BSD license
//
// Using this required having the $JAVA_HOME and $HADOOP_HOME environment
// variables set, so that libjvm and libhdfs can be located easily
// Copyright (C) 2015 Dato, Inc.
// All rights reserved.
//
// This software may be modified and distributed under the terms
// of the BSD license. See the LICENSE file for details.
#ifdef HAS_HADOOP
#ifndef _WIN32
#include <dlfcn.h>
#else
#include <windows.h>
#include <winsock2.h>
// TODO(wesm): address when/if we add windows support
// #include <util/syserr_reporting.hpp>
#endif
extern "C" {
#include <hdfs.h>
}
#include <iostream>
#include <mutex>
#include <sstream>
#include <string>
#include <type_traits>
#include <vector>
#include <boost/filesystem.hpp> // NOLINT
#include "arrow/io/hdfs-internal.h"
#include "arrow/status.h"
#include "arrow/util/visibility.h"
namespace fs = boost::filesystem;
#ifndef _WIN32
static void* libjvm_handle = NULL;
#else
static HINSTANCE libjvm_handle = NULL;
#endif
/*
* All the shim pointers
*/
// Helper functions for dlopens
static std::vector<fs::path> get_potential_libjvm_paths();
static std::vector<fs::path> get_potential_libhdfs_paths();
static std::vector<fs::path> get_potential_libhdfs3_paths();
static arrow::Status try_dlopen(std::vector<fs::path> potential_paths, const char* name,
#ifndef _WIN32
void*& out_handle);
#else
HINSTANCE& out_handle);
#endif
static std::vector<fs::path> get_potential_libhdfs_paths() {
std::vector<fs::path> libhdfs_potential_paths;
std::string file_name;
// OS-specific file name
#ifdef __WIN32
file_name = "hdfs.dll";
#elif __APPLE__
file_name = "libhdfs.dylib";
#else
file_name = "libhdfs.so";
#endif
// Common paths
std::vector<fs::path> search_paths = {fs::path(""), fs::path(".")};
// Path from environment variable
const char* hadoop_home = std::getenv("HADOOP_HOME");
if (hadoop_home != nullptr) {
auto path = fs::path(hadoop_home) / "lib/native";
search_paths.push_back(path);
}
const char* libhdfs_dir = std::getenv("ARROW_LIBHDFS_DIR");
if (libhdfs_dir != nullptr) { search_paths.push_back(fs::path(libhdfs_dir)); }
// All paths with file name
for (auto& path : search_paths) {
libhdfs_potential_paths.push_back(path / file_name);
}
return libhdfs_potential_paths;
}
static std::vector<fs::path> get_potential_libhdfs3_paths() {
std::vector<fs::path> potential_paths;
std::string file_name;
// OS-specific file name
#ifdef __WIN32
file_name = "hdfs3.dll";
#elif __APPLE__
file_name = "libhdfs3.dylib";
#else
file_name = "libhdfs3.so";
#endif
// Common paths
std::vector<fs::path> search_paths = {fs::path(""), fs::path(".")};
const char* libhdfs3_dir = std::getenv("ARROW_LIBHDFS3_DIR");
if (libhdfs3_dir != nullptr) { search_paths.push_back(fs::path(libhdfs3_dir)); }
// All paths with file name
for (auto& path : search_paths) {
potential_paths.push_back(path / file_name);
}
return potential_paths;
}
static std::vector<fs::path> get_potential_libjvm_paths() {
std::vector<fs::path> libjvm_potential_paths;
std::vector<fs::path> search_prefixes;
std::vector<fs::path> search_suffixes;
std::string file_name;
// From heuristics
#ifdef __WIN32
search_prefixes = {""};
search_suffixes = {"/jre/bin/server", "/bin/server"};
file_name = "jvm.dll";
#elif __APPLE__
search_prefixes = {""};
search_suffixes = {"", "/jre/lib/server"};
file_name = "libjvm.dylib";
// SFrame uses /usr/libexec/java_home to find JAVA_HOME; for now we are
// expecting users to set an environment variable
#else
search_prefixes = {
"/usr/lib/jvm/default-java", // ubuntu / debian distros
"/usr/lib/jvm/java", // rhel6
"/usr/lib/jvm", // centos6
"/usr/lib64/jvm", // opensuse 13
"/usr/local/lib/jvm/default-java", // alt ubuntu / debian distros
"/usr/local/lib/jvm/java", // alt rhel6
"/usr/local/lib/jvm", // alt centos6
"/usr/local/lib64/jvm", // alt opensuse 13
"/usr/local/lib/jvm/java-7-openjdk-amd64", // alt ubuntu / debian distros
"/usr/lib/jvm/java-7-openjdk-amd64", // alt ubuntu / debian distros
"/usr/local/lib/jvm/java-6-openjdk-amd64", // alt ubuntu / debian distros
"/usr/lib/jvm/java-6-openjdk-amd64", // alt ubuntu / debian distros
"/usr/lib/jvm/java-7-oracle", // alt ubuntu
"/usr/lib/jvm/java-8-oracle", // alt ubuntu
"/usr/lib/jvm/java-6-oracle", // alt ubuntu
"/usr/local/lib/jvm/java-7-oracle", // alt ubuntu
"/usr/local/lib/jvm/java-8-oracle", // alt ubuntu
"/usr/local/lib/jvm/java-6-oracle", // alt ubuntu
"/usr/lib/jvm/default", // alt centos
"/usr/java/latest", // alt centos
};
search_suffixes = {"/jre/lib/amd64/server"};
file_name = "libjvm.so";
#endif
// From direct environment variable
char* env_value = NULL;
if ((env_value = getenv("JAVA_HOME")) != NULL) {
search_prefixes.insert(search_prefixes.begin(), env_value);
}
// Generate cross product between search_prefixes, search_suffixes, and file_name
for (auto& prefix : search_prefixes) {
for (auto& suffix : search_suffixes) {
auto path = (fs::path(prefix) / fs::path(suffix) / fs::path(file_name));
libjvm_potential_paths.push_back(path);
}
}
return libjvm_potential_paths;
}
#ifndef _WIN32
static arrow::Status try_dlopen(
std::vector<fs::path> potential_paths, const char* name, void*& out_handle) {
std::vector<std::string> error_messages;
for (auto& i : potential_paths) {
i.make_preferred();
out_handle = dlopen(i.native().c_str(), RTLD_NOW | RTLD_LOCAL);
if (out_handle != NULL) {
// std::cout << "Loaded " << i << std::endl;
break;
} else {
const char* err_msg = dlerror();
if (err_msg != NULL) {
error_messages.push_back(std::string(err_msg));
} else {
error_messages.push_back(std::string(" returned NULL"));
}
}
}
if (out_handle == NULL) {
std::stringstream ss;
ss << "Unable to load " << name;
return arrow::Status::IOError(ss.str());
}
return arrow::Status::OK();
}
#else
static arrow::Status try_dlopen(
std::vector<fs::path> potential_paths, const char* name, HINSTANCE& out_handle) {
std::vector<std::string> error_messages;
for (auto& i : potential_paths) {
i.make_preferred();
out_handle = LoadLibrary(i.string().c_str());
if (out_handle != NULL) {
break;
} else {
// error_messages.push_back(get_last_err_str(GetLastError()));
}
}
if (out_handle == NULL) {
std::stringstream ss;
ss << "Unable to load " << name;
return arrow::Status::IOError(ss.str());
}
return arrow::Status::OK();
}
#endif // _WIN32
static inline void* GetLibrarySymbol(void* handle, const char* symbol) {
if (handle == NULL) return NULL;
#ifndef _WIN32
return dlsym(handle, symbol);
#else
void* ret = reinterpret_cast<void*>(GetProcAddress(handle, symbol));
if (ret == NULL) {
// logstream(LOG_INFO) << "GetProcAddress error: "
// << get_last_err_str(GetLastError()) << std::endl;
}
return ret;
#endif
}
#define GET_SYMBOL_REQUIRED(SHIM, SYMBOL_NAME) \
do { \
if (!SHIM->SYMBOL_NAME) { \
*reinterpret_cast<void**>(&SHIM->SYMBOL_NAME) = \
GetLibrarySymbol(SHIM->handle, "" #SYMBOL_NAME); \
} \
if (!SHIM->SYMBOL_NAME) \
return Status::IOError("Getting symbol " #SYMBOL_NAME "failed"); \
} while (0)
#define GET_SYMBOL(SHIM, SYMBOL_NAME) \
if (!SHIM->SYMBOL_NAME) { \
*reinterpret_cast<void**>(&SHIM->SYMBOL_NAME) = \
GetLibrarySymbol(SHIM->handle, "" #SYMBOL_NAME); \
}
namespace arrow {
namespace io {
static LibHdfsShim libhdfs_shim;
static LibHdfsShim libhdfs3_shim;
hdfsBuilder* LibHdfsShim::NewBuilder(void) {
return this->hdfsNewBuilder();
}
void LibHdfsShim::BuilderSetNameNode(hdfsBuilder* bld, const char* nn) {
this->hdfsBuilderSetNameNode(bld, nn);
}
void LibHdfsShim::BuilderSetNameNodePort(hdfsBuilder* bld, tPort port) {
this->hdfsBuilderSetNameNodePort(bld, port);
}
void LibHdfsShim::BuilderSetUserName(hdfsBuilder* bld, const char* userName) {
this->hdfsBuilderSetUserName(bld, userName);
}
void LibHdfsShim::BuilderSetKerbTicketCachePath(
hdfsBuilder* bld, const char* kerbTicketCachePath) {
this->hdfsBuilderSetKerbTicketCachePath(bld, kerbTicketCachePath);
}
hdfsFS LibHdfsShim::BuilderConnect(hdfsBuilder* bld) {
return this->hdfsBuilderConnect(bld);
}
int LibHdfsShim::Disconnect(hdfsFS fs) {
return this->hdfsDisconnect(fs);
}
hdfsFile LibHdfsShim::OpenFile(hdfsFS fs, const char* path, int flags, int bufferSize,
short replication, tSize blocksize) { // NOLINT
return this->hdfsOpenFile(fs, path, flags, bufferSize, replication, blocksize);
}
int LibHdfsShim::CloseFile(hdfsFS fs, hdfsFile file) {
return this->hdfsCloseFile(fs, file);
}
int LibHdfsShim::Exists(hdfsFS fs, const char* path) {
return this->hdfsExists(fs, path);
}
int LibHdfsShim::Seek(hdfsFS fs, hdfsFile file, tOffset desiredPos) {
return this->hdfsSeek(fs, file, desiredPos);
}
tOffset LibHdfsShim::Tell(hdfsFS fs, hdfsFile file) {
return this->hdfsTell(fs, file);
}
tSize LibHdfsShim::Read(hdfsFS fs, hdfsFile file, void* buffer, tSize length) {
return this->hdfsRead(fs, file, buffer, length);
}
bool LibHdfsShim::HasPread() {
GET_SYMBOL(this, hdfsPread);
return this->hdfsPread != nullptr;
}
tSize LibHdfsShim::Pread(
hdfsFS fs, hdfsFile file, tOffset position, void* buffer, tSize length) {
GET_SYMBOL(this, hdfsPread);
return this->hdfsPread(fs, file, position, buffer, length);
}
tSize LibHdfsShim::Write(hdfsFS fs, hdfsFile file, const void* buffer, tSize length) {
return this->hdfsWrite(fs, file, buffer, length);
}
int LibHdfsShim::Flush(hdfsFS fs, hdfsFile file) {
return this->hdfsFlush(fs, file);
}
int LibHdfsShim::Available(hdfsFS fs, hdfsFile file) {
GET_SYMBOL(this, hdfsAvailable);
if (this->hdfsAvailable)
return this->hdfsAvailable(fs, file);
else
return 0;
}
int LibHdfsShim::Copy(hdfsFS srcFS, const char* src, hdfsFS dstFS, const char* dst) {
GET_SYMBOL(this, hdfsCopy);
if (this->hdfsCopy)
return this->hdfsCopy(srcFS, src, dstFS, dst);
else
return 0;
}
int LibHdfsShim::Move(hdfsFS srcFS, const char* src, hdfsFS dstFS, const char* dst) {
GET_SYMBOL(this, hdfsMove);
if (this->hdfsMove)
return this->hdfsMove(srcFS, src, dstFS, dst);
else
return 0;
}
int LibHdfsShim::Delete(hdfsFS fs, const char* path, int recursive) {
return this->hdfsDelete(fs, path, recursive);
}
int LibHdfsShim::Rename(hdfsFS fs, const char* oldPath, const char* newPath) {
GET_SYMBOL(this, hdfsRename);
if (this->hdfsRename)
return this->hdfsRename(fs, oldPath, newPath);
else
return 0;
}
char* LibHdfsShim::GetWorkingDirectory(hdfsFS fs, char* buffer, size_t bufferSize) {
GET_SYMBOL(this, hdfsGetWorkingDirectory);
if (this->hdfsGetWorkingDirectory) {
return this->hdfsGetWorkingDirectory(fs, buffer, bufferSize);
} else {
return NULL;
}
}
int LibHdfsShim::SetWorkingDirectory(hdfsFS fs, const char* path) {
GET_SYMBOL(this, hdfsSetWorkingDirectory);
if (this->hdfsSetWorkingDirectory) {
return this->hdfsSetWorkingDirectory(fs, path);
} else {
return 0;
}
}
int LibHdfsShim::CreateDirectory(hdfsFS fs, const char* path) {
return this->hdfsCreateDirectory(fs, path);
}
int LibHdfsShim::SetReplication(hdfsFS fs, const char* path, int16_t replication) {
GET_SYMBOL(this, hdfsSetReplication);
if (this->hdfsSetReplication) {
return this->hdfsSetReplication(fs, path, replication);
} else {
return 0;
}
}
hdfsFileInfo* LibHdfsShim::ListDirectory(hdfsFS fs, const char* path, int* numEntries) {
return this->hdfsListDirectory(fs, path, numEntries);
}
hdfsFileInfo* LibHdfsShim::GetPathInfo(hdfsFS fs, const char* path) {
return this->hdfsGetPathInfo(fs, path);
}
void LibHdfsShim::FreeFileInfo(hdfsFileInfo* hdfsFileInfo, int numEntries) {
this->hdfsFreeFileInfo(hdfsFileInfo, numEntries);
}
char*** LibHdfsShim::GetHosts(
hdfsFS fs, const char* path, tOffset start, tOffset length) {
GET_SYMBOL(this, hdfsGetHosts);
if (this->hdfsGetHosts) {
return this->hdfsGetHosts(fs, path, start, length);
} else {
return NULL;
}
}
void LibHdfsShim::FreeHosts(char*** blockHosts) {
GET_SYMBOL(this, hdfsFreeHosts);
if (this->hdfsFreeHosts) { this->hdfsFreeHosts(blockHosts); }
}
tOffset LibHdfsShim::GetDefaultBlockSize(hdfsFS fs) {
GET_SYMBOL(this, hdfsGetDefaultBlockSize);
if (this->hdfsGetDefaultBlockSize) {
return this->hdfsGetDefaultBlockSize(fs);
} else {
return 0;
}
}
tOffset LibHdfsShim::GetCapacity(hdfsFS fs) {
return this->hdfsGetCapacity(fs);
}
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;
}
}
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;
}
}
int LibHdfsShim::Utime(hdfsFS fs, const char* path, tTime mtime, tTime atime) {
GET_SYMBOL(this, hdfsUtime);
if (this->hdfsUtime) {
return this->hdfsUtime(fs, path, mtime, atime);
} else {
return 0;
}
}
Status LibHdfsShim::GetRequiredSymbols() {
GET_SYMBOL_REQUIRED(this, hdfsNewBuilder);
GET_SYMBOL_REQUIRED(this, hdfsBuilderSetNameNode);
GET_SYMBOL_REQUIRED(this, hdfsBuilderSetNameNodePort);
GET_SYMBOL_REQUIRED(this, hdfsBuilderSetUserName);
GET_SYMBOL_REQUIRED(this, hdfsBuilderSetKerbTicketCachePath);
GET_SYMBOL_REQUIRED(this, hdfsBuilderConnect);
GET_SYMBOL_REQUIRED(this, hdfsCreateDirectory);
GET_SYMBOL_REQUIRED(this, hdfsDelete);
GET_SYMBOL_REQUIRED(this, hdfsDisconnect);
GET_SYMBOL_REQUIRED(this, hdfsExists);
GET_SYMBOL_REQUIRED(this, hdfsFreeFileInfo);
GET_SYMBOL_REQUIRED(this, hdfsGetCapacity);
GET_SYMBOL_REQUIRED(this, hdfsGetUsed);
GET_SYMBOL_REQUIRED(this, hdfsGetPathInfo);
GET_SYMBOL_REQUIRED(this, hdfsListDirectory);
// File methods
GET_SYMBOL_REQUIRED(this, hdfsCloseFile);
GET_SYMBOL_REQUIRED(this, hdfsFlush);
GET_SYMBOL_REQUIRED(this, hdfsOpenFile);
GET_SYMBOL_REQUIRED(this, hdfsRead);
GET_SYMBOL_REQUIRED(this, hdfsSeek);
GET_SYMBOL_REQUIRED(this, hdfsTell);
GET_SYMBOL_REQUIRED(this, hdfsWrite);
return Status::OK();
}
Status ARROW_EXPORT ConnectLibHdfs(LibHdfsShim** driver) {
static std::mutex lock;
std::lock_guard<std::mutex> guard(lock);
LibHdfsShim* shim = &libhdfs_shim;
static bool shim_attempted = false;
if (!shim_attempted) {
shim_attempted = true;
shim->Initialize();
std::vector<fs::path> libjvm_potential_paths = get_potential_libjvm_paths();
RETURN_NOT_OK(try_dlopen(libjvm_potential_paths, "libjvm", libjvm_handle));
std::vector<fs::path> libhdfs_potential_paths = get_potential_libhdfs_paths();
RETURN_NOT_OK(try_dlopen(libhdfs_potential_paths, "libhdfs", shim->handle));
} else if (shim->handle == nullptr) {
return Status::IOError("Prior attempt to load libhdfs failed");
}
*driver = shim;
return shim->GetRequiredSymbols();
}
Status ARROW_EXPORT ConnectLibHdfs3(LibHdfsShim** driver) {
static std::mutex lock;
std::lock_guard<std::mutex> guard(lock);
LibHdfsShim* shim = &libhdfs3_shim;
static bool shim_attempted = false;
if (!shim_attempted) {
shim_attempted = true;
shim->Initialize();
std::vector<fs::path> libhdfs3_potential_paths = get_potential_libhdfs3_paths();
RETURN_NOT_OK(try_dlopen(libhdfs3_potential_paths, "libhdfs3", shim->handle));
} else if (shim->handle == nullptr) {
return Status::IOError("Prior attempt to load libhdfs3 failed");
}
*driver = shim;
return shim->GetRequiredSymbols();
}
} // namespace io
} // namespace arrow
#endif // HAS_HADOOP
cpp/src/arrow/io/hdfs-internal.h
+203
View File
@@ -0,0 +1,203 @@
// 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.
#ifndef ARROW_IO_HDFS_INTERNAL
#define ARROW_IO_HDFS_INTERNAL
#include <hdfs.h>
namespace arrow {
class Status;
namespace io {
// NOTE(wesm): cpplint does not like use of short and other imprecise C types
struct LibHdfsShim {
#ifndef _WIN32
void* handle;
#else
HINSTANCE handle;
#endif
hdfsBuilder* (*hdfsNewBuilder)(void);
void (*hdfsBuilderSetNameNode)(hdfsBuilder* bld, const char* nn);
void (*hdfsBuilderSetNameNodePort)(hdfsBuilder* bld, tPort port);
void (*hdfsBuilderSetUserName)(hdfsBuilder* bld, const char* userName);
void (*hdfsBuilderSetKerbTicketCachePath)(
hdfsBuilder* bld, const char* kerbTicketCachePath);
hdfsFS (*hdfsBuilderConnect)(hdfsBuilder* bld);
int (*hdfsDisconnect)(hdfsFS fs);
hdfsFile (*hdfsOpenFile)(hdfsFS fs, const char* path, int flags, int bufferSize,
short replication, tSize blocksize); // NOLINT
int (*hdfsCloseFile)(hdfsFS fs, hdfsFile file);
int (*hdfsExists)(hdfsFS fs, const char* path);
int (*hdfsSeek)(hdfsFS fs, hdfsFile file, tOffset desiredPos);
tOffset (*hdfsTell)(hdfsFS fs, hdfsFile file);
tSize (*hdfsRead)(hdfsFS fs, hdfsFile file, void* buffer, tSize length);
tSize (*hdfsPread)(
hdfsFS fs, hdfsFile file, tOffset position, void* buffer, tSize length);
tSize (*hdfsWrite)(hdfsFS fs, hdfsFile file, const void* buffer, tSize length);
int (*hdfsFlush)(hdfsFS fs, hdfsFile file);
int (*hdfsAvailable)(hdfsFS fs, hdfsFile file);
int (*hdfsCopy)(hdfsFS srcFS, const char* src, hdfsFS dstFS, const char* dst);
int (*hdfsMove)(hdfsFS srcFS, const char* src, hdfsFS dstFS, const char* dst);
int (*hdfsDelete)(hdfsFS fs, const char* path, int recursive);
int (*hdfsRename)(hdfsFS fs, const char* oldPath, const char* newPath);
char* (*hdfsGetWorkingDirectory)(hdfsFS fs, char* buffer, size_t bufferSize);
int (*hdfsSetWorkingDirectory)(hdfsFS fs, const char* path);
int (*hdfsCreateDirectory)(hdfsFS fs, const char* path);
int (*hdfsSetReplication)(hdfsFS fs, const char* path, int16_t replication);
hdfsFileInfo* (*hdfsListDirectory)(hdfsFS fs, const char* path, int* numEntries);
hdfsFileInfo* (*hdfsGetPathInfo)(hdfsFS fs, const char* path);
void (*hdfsFreeFileInfo)(hdfsFileInfo* hdfsFileInfo, int numEntries);
char*** (*hdfsGetHosts)(hdfsFS fs, const char* path, tOffset start, tOffset length);
void (*hdfsFreeHosts)(char*** blockHosts);
tOffset (*hdfsGetDefaultBlockSize)(hdfsFS fs);
tOffset (*hdfsGetCapacity)(hdfsFS fs);
tOffset (*hdfsGetUsed)(hdfsFS fs);
int (*hdfsChown)(hdfsFS fs, const char* path, const char* owner, const char* group);
int (*hdfsChmod)(hdfsFS fs, const char* path, short mode); // NOLINT
int (*hdfsUtime)(hdfsFS fs, const char* path, tTime mtime, tTime atime);
void Initialize() {
this->handle = nullptr;
this->hdfsNewBuilder = nullptr;
this->hdfsBuilderSetNameNode = nullptr;
this->hdfsBuilderSetNameNodePort = nullptr;
this->hdfsBuilderSetUserName = nullptr;
this->hdfsBuilderSetKerbTicketCachePath = nullptr;
this->hdfsBuilderConnect = nullptr;
this->hdfsDisconnect = nullptr;
this->hdfsOpenFile = nullptr;
this->hdfsCloseFile = nullptr;
this->hdfsExists = nullptr;
this->hdfsSeek = nullptr;
this->hdfsTell = nullptr;
this->hdfsRead = nullptr;
this->hdfsPread = nullptr;
this->hdfsWrite = nullptr;
this->hdfsFlush = nullptr;
this->hdfsAvailable = nullptr;
this->hdfsCopy = nullptr;
this->hdfsMove = nullptr;
this->hdfsDelete = nullptr;
this->hdfsRename = nullptr;
this->hdfsGetWorkingDirectory = nullptr;
this->hdfsSetWorkingDirectory = nullptr;
this->hdfsCreateDirectory = nullptr;
this->hdfsSetReplication = nullptr;
this->hdfsListDirectory = nullptr;
this->hdfsGetPathInfo = nullptr;
this->hdfsFreeFileInfo = nullptr;
this->hdfsGetHosts = nullptr;
this->hdfsFreeHosts = nullptr;
this->hdfsGetDefaultBlockSize = nullptr;
this->hdfsGetCapacity = nullptr;
this->hdfsGetUsed = nullptr;
this->hdfsChown = nullptr;
this->hdfsChmod = nullptr;
this->hdfsUtime = nullptr;
}
hdfsBuilder* NewBuilder(void);
void BuilderSetNameNode(hdfsBuilder* bld, const char* nn);
void BuilderSetNameNodePort(hdfsBuilder* bld, tPort port);
void BuilderSetUserName(hdfsBuilder* bld, const char* userName);
void BuilderSetKerbTicketCachePath(hdfsBuilder* bld, const char* kerbTicketCachePath);
hdfsFS BuilderConnect(hdfsBuilder* bld);
int Disconnect(hdfsFS fs);
hdfsFile OpenFile(hdfsFS fs, const char* path, int flags, int bufferSize,
short replication, tSize blocksize); // NOLINT
int CloseFile(hdfsFS fs, hdfsFile file);
int Exists(hdfsFS fs, const char* path);
int Seek(hdfsFS fs, hdfsFile file, tOffset desiredPos);
tOffset Tell(hdfsFS fs, hdfsFile file);
tSize Read(hdfsFS fs, hdfsFile file, void* buffer, tSize length);
bool HasPread();
tSize Pread(hdfsFS fs, hdfsFile file, tOffset position, void* buffer, tSize length);
tSize Write(hdfsFS fs, hdfsFile file, const void* buffer, tSize length);
int Flush(hdfsFS fs, hdfsFile file);
int Available(hdfsFS fs, hdfsFile file);
int Copy(hdfsFS srcFS, const char* src, hdfsFS dstFS, const char* dst);
int Move(hdfsFS srcFS, const char* src, hdfsFS dstFS, const char* dst);
int Delete(hdfsFS fs, const char* path, int recursive);
int Rename(hdfsFS fs, const char* oldPath, const char* newPath);
char* GetWorkingDirectory(hdfsFS fs, char* buffer, size_t bufferSize);
int SetWorkingDirectory(hdfsFS fs, const char* path);
int CreateDirectory(hdfsFS fs, const char* path);
int SetReplication(hdfsFS fs, const char* path, int16_t replication);
hdfsFileInfo* ListDirectory(hdfsFS fs, const char* path, int* numEntries);
hdfsFileInfo* GetPathInfo(hdfsFS fs, const char* path);
void FreeFileInfo(hdfsFileInfo* hdfsFileInfo, int numEntries);
char*** GetHosts(hdfsFS fs, const char* path, tOffset start, tOffset length);
void FreeHosts(char*** blockHosts);
tOffset GetDefaultBlockSize(hdfsFS fs);
tOffset GetCapacity(hdfsFS fs);
tOffset GetUsed(hdfsFS fs);
int Chown(hdfsFS fs, const char* path, const char* owner, const char* group);
int Chmod(hdfsFS fs, const char* path, short mode); // NOLINT
int Utime(hdfsFS fs, const char* path, tTime mtime, tTime atime);
Status GetRequiredSymbols();
};
Status ConnectLibHdfs(LibHdfsShim** driver);
Status ConnectLibHdfs3(LibHdfsShim** driver);
} // namespace io
} // namespace arrow
#endif // ARROW_IO_HDFS_INTERNAL
cpp/src/arrow/io/hdfs.cc
+69 -35
View File
@@ -23,6 +23,7 @@
#include <string>
#include "arrow/buffer.h"
#include "arrow/io/hdfs-internal.h"
#include "arrow/io/hdfs.h"
#include "arrow/memory_pool.h"
#include "arrow/status.h"
@@ -59,21 +60,23 @@ static constexpr int kDefaultHdfsBufferSize = 1 << 16;
class HdfsAnyFileImpl {
public:
void set_members(const std::string& path, hdfsFS fs, hdfsFile handle) {
void set_members(
const std::string& path, LibHdfsShim* driver, hdfsFS fs, hdfsFile handle) {
path_ = path;
driver_ = driver;
fs_ = fs;
file_ = handle;
is_open_ = true;
}
Status Seek(int64_t position) {
int ret = hdfsSeek(fs_, file_, position);
int ret = driver_->Seek(fs_, file_, position);
CHECK_FAILURE(ret, "seek");
return Status::OK();
}
Status Tell(int64_t* offset) {
int64_t ret = hdfsTell(fs_, file_);
int64_t ret = driver_->Tell(fs_, file_);
CHECK_FAILURE(ret, "tell");
*offset = ret;
return Status::OK();
@@ -84,6 +87,8 @@ class HdfsAnyFileImpl {
protected:
std::string path_;
LibHdfsShim* driver_;
// These are pointers in libhdfs, so OK to copy
hdfsFS fs_;
hdfsFile file_;
@@ -98,7 +103,7 @@ class HdfsReadableFile::HdfsReadableFileImpl : public HdfsAnyFileImpl {
Status Close() {
if (is_open_) {
int ret = hdfsCloseFile(fs_, file_);
int ret = driver_->CloseFile(fs_, file_);
CHECK_FAILURE(ret, "CloseFile");
is_open_ = false;
}
@@ -106,8 +111,14 @@ class HdfsReadableFile::HdfsReadableFileImpl : public HdfsAnyFileImpl {
}
Status ReadAt(int64_t position, int64_t nbytes, int64_t* bytes_read, uint8_t* buffer) {
tSize ret = hdfsPread(fs_, file_, static_cast<tOffset>(position),
reinterpret_cast<void*>(buffer), nbytes);
tSize ret;
if (driver_->HasPread()) {
ret = driver_->Pread(fs_, file_, static_cast<tOffset>(position),
reinterpret_cast<void*>(buffer), nbytes);
} else {
RETURN_NOT_OK(Seek(position));
return Read(nbytes, bytes_read, buffer);
}
RETURN_NOT_OK(CheckReadResult(ret));
*bytes_read = ret;
return Status::OK();
@@ -129,7 +140,7 @@ class HdfsReadableFile::HdfsReadableFileImpl : public HdfsAnyFileImpl {
Status Read(int64_t nbytes, int64_t* bytes_read, uint8_t* buffer) {
int64_t total_bytes = 0;
while (total_bytes < nbytes) {
tSize ret = hdfsRead(fs_, file_, reinterpret_cast<void*>(buffer + total_bytes),
tSize ret = driver_->Read(fs_, file_, reinterpret_cast<void*>(buffer + total_bytes),
std::min<int64_t>(buffer_size_, nbytes - total_bytes));
RETURN_NOT_OK(CheckReadResult(ret));
total_bytes += ret;
@@ -153,11 +164,11 @@ class HdfsReadableFile::HdfsReadableFileImpl : public HdfsAnyFileImpl {
}
Status GetSize(int64_t* size) {
hdfsFileInfo* entry = hdfsGetPathInfo(fs_, path_.c_str());
hdfsFileInfo* entry = driver_->GetPathInfo(fs_, path_.c_str());
if (entry == nullptr) { return Status::IOError("HDFS: GetPathInfo failed"); }
*size = entry->mSize;
hdfsFreeFileInfo(entry, 1);
driver_->FreeFileInfo(entry, 1);
return Status::OK();
}
@@ -227,9 +238,9 @@ class HdfsOutputStream::HdfsOutputStreamImpl : public HdfsAnyFileImpl {
Status Close() {
if (is_open_) {
int ret = hdfsFlush(fs_, file_);
int ret = driver_->Flush(fs_, file_);
CHECK_FAILURE(ret, "Flush");
ret = hdfsCloseFile(fs_, file_);
ret = driver_->CloseFile(fs_, file_);
CHECK_FAILURE(ret, "CloseFile");
is_open_ = false;
}
@@ -237,7 +248,7 @@ class HdfsOutputStream::HdfsOutputStreamImpl : public HdfsAnyFileImpl {
}
Status Write(const uint8_t* buffer, int64_t nbytes, int64_t* bytes_written) {
tSize ret = hdfsWrite(fs_, file_, reinterpret_cast<const void*>(buffer), nbytes);
tSize ret = driver_->Write(fs_, file_, reinterpret_cast<const void*>(buffer), nbytes);
CHECK_FAILURE(ret, "Write");
*bytes_written = ret;
return Status::OK();
@@ -297,17 +308,25 @@ class HdfsClient::HdfsClientImpl {
HdfsClientImpl() {}
Status Connect(const HdfsConnectionConfig* config) {
RETURN_NOT_OK(ConnectLibHdfs());
if (config->driver == HdfsDriver::LIBHDFS3) {
RETURN_NOT_OK(ConnectLibHdfs3(&driver_));
} else {
RETURN_NOT_OK(ConnectLibHdfs(&driver_));
}
// connect to HDFS with the builder object
hdfsBuilder* builder = hdfsNewBuilder();
if (!config->host.empty()) { hdfsBuilderSetNameNode(builder, config->host.c_str()); }
hdfsBuilderSetNameNodePort(builder, config->port);
if (!config->user.empty()) { hdfsBuilderSetUserName(builder, config->user.c_str()); }
if (!config->kerb_ticket.empty()) {
hdfsBuilderSetKerbTicketCachePath(builder, config->kerb_ticket.c_str());
hdfsBuilder* builder = driver_->NewBuilder();
if (!config->host.empty()) {
driver_->BuilderSetNameNode(builder, config->host.c_str());
}
fs_ = hdfsBuilderConnect(builder);
driver_->BuilderSetNameNodePort(builder, config->port);
if (!config->user.empty()) {
driver_->BuilderSetUserName(builder, config->user.c_str());
}
if (!config->kerb_ticket.empty()) {
driver_->BuilderSetKerbTicketCachePath(builder, config->kerb_ticket.c_str());
}
fs_ = driver_->BuilderConnect(builder);
if (fs_ == nullptr) { return Status::IOError("HDFS connection failed"); }
namenode_host_ = config->host;
@@ -319,19 +338,19 @@ class HdfsClient::HdfsClientImpl {
}
Status CreateDirectory(const std::string& path) {
int ret = hdfsCreateDirectory(fs_, path.c_str());
int ret = driver_->CreateDirectory(fs_, path.c_str());
CHECK_FAILURE(ret, "create directory");
return Status::OK();
}
Status Delete(const std::string& path, bool recursive) {
int ret = hdfsDelete(fs_, path.c_str(), static_cast<int>(recursive));
int ret = driver_->Delete(fs_, path.c_str(), static_cast<int>(recursive));
CHECK_FAILURE(ret, "delete");
return Status::OK();
}
Status Disconnect() {
int ret = hdfsDisconnect(fs_);
int ret = driver_->Disconnect(fs_);
CHECK_FAILURE(ret, "hdfsFS::Disconnect");
return Status::OK();
}
@@ -339,38 +358,38 @@ class HdfsClient::HdfsClientImpl {
bool Exists(const std::string& path) {
// hdfsExists does not distinguish between RPC failure and the file not
// existing
int ret = hdfsExists(fs_, path.c_str());
int ret = driver_->Exists(fs_, path.c_str());
return ret == 0;
}
Status GetCapacity(int64_t* nbytes) {
tOffset ret = hdfsGetCapacity(fs_);
tOffset ret = driver_->GetCapacity(fs_);
CHECK_FAILURE(ret, "GetCapacity");
*nbytes = ret;
return Status::OK();
}
Status GetUsed(int64_t* nbytes) {
tOffset ret = hdfsGetUsed(fs_);
tOffset ret = driver_->GetUsed(fs_);
CHECK_FAILURE(ret, "GetUsed");
*nbytes = ret;
return Status::OK();
}
Status GetPathInfo(const std::string& path, HdfsPathInfo* info) {
hdfsFileInfo* entry = hdfsGetPathInfo(fs_, path.c_str());
hdfsFileInfo* entry = driver_->GetPathInfo(fs_, path.c_str());
if (entry == nullptr) { return Status::IOError("HDFS: GetPathInfo failed"); }
SetPathInfo(entry, info);
hdfsFreeFileInfo(entry, 1);
driver_->FreeFileInfo(entry, 1);
return Status::OK();
}
Status ListDirectory(const std::string& path, std::vector<HdfsPathInfo>* listing) {
int num_entries = 0;
hdfsFileInfo* entries = hdfsListDirectory(fs_, path.c_str(), &num_entries);
hdfsFileInfo* entries = driver_->ListDirectory(fs_, path.c_str(), &num_entries);
if (entries == nullptr) {
// If the directory is empty, entries is NULL but errno is 0. Non-zero
@@ -391,14 +410,14 @@ class HdfsClient::HdfsClientImpl {
}
// Free libhdfs file info
hdfsFreeFileInfo(entries, num_entries);
driver_->FreeFileInfo(entries, num_entries);
return Status::OK();
}
Status OpenReadable(const std::string& path, int32_t buffer_size,
std::shared_ptr<HdfsReadableFile>* file) {
hdfsFile handle = hdfsOpenFile(fs_, path.c_str(), O_RDONLY, buffer_size, 0, 0);
hdfsFile handle = driver_->OpenFile(fs_, path.c_str(), O_RDONLY, buffer_size, 0, 0);
if (handle == nullptr) {
// TODO(wesm): determine cause of failure
@@ -409,7 +428,7 @@ class HdfsClient::HdfsClientImpl {
// std::make_shared does not work with private ctors
*file = std::shared_ptr<HdfsReadableFile>(new HdfsReadableFile());
(*file)->impl_->set_members(path, fs_, handle);
(*file)->impl_->set_members(path, driver_, fs_, handle);
(*file)->impl_->set_buffer_size(buffer_size);
return Status::OK();
@@ -421,7 +440,7 @@ class HdfsClient::HdfsClientImpl {
int flags = O_WRONLY;
if (append) flags |= O_APPEND;
hdfsFile handle = hdfsOpenFile(
hdfsFile handle = driver_->OpenFile(
fs_, path.c_str(), flags, buffer_size, replication, default_block_size);
if (handle == nullptr) {
@@ -433,18 +452,20 @@ class HdfsClient::HdfsClientImpl {
// std::make_shared does not work with private ctors
*file = std::shared_ptr<HdfsOutputStream>(new HdfsOutputStream());
(*file)->impl_->set_members(path, fs_, handle);
(*file)->impl_->set_members(path, driver_, fs_, handle);
return Status::OK();
}
Status Rename(const std::string& src, const std::string& dst) {
int ret = hdfsRename(fs_, src.c_str(), dst.c_str());
int ret = driver_->Rename(fs_, src.c_str(), dst.c_str());
CHECK_FAILURE(ret, "Rename");
return Status::OK();
}
private:
LibHdfsShim* driver_;
std::string namenode_host_;
std::string user_;
int port_;
@@ -530,5 +551,18 @@ Status HdfsClient::Rename(const std::string& src, const std::string& dst) {
return impl_->Rename(src, dst);
}
// ----------------------------------------------------------------------
// Allow public API users to check whether we are set up correctly
Status HaveLibHdfs() {
LibHdfsShim* driver;
return ConnectLibHdfs(&driver);
}
Status HaveLibHdfs3() {
LibHdfsShim* driver;
return ConnectLibHdfs3(&driver);
}
} // namespace io
} // namespace arrow
cpp/src/arrow/io/hdfs.h
+5 -1
View File
@@ -56,11 +56,14 @@ struct HdfsPathInfo {
int16_t permissions;
};
enum class HdfsDriver : char { LIBHDFS, LIBHDFS3 };
struct HdfsConnectionConfig {
std::string host;
int port;
std::string user;
std::string kerb_ticket;
HdfsDriver driver;
};
class ARROW_EXPORT HdfsClient : public FileSystemClient {
@@ -218,7 +221,8 @@ class ARROW_EXPORT HdfsOutputStream : public OutputStream {
DISALLOW_COPY_AND_ASSIGN(HdfsOutputStream);
};
Status ARROW_EXPORT ConnectLibHdfs();
Status ARROW_EXPORT HaveLibHdfs();
Status ARROW_EXPORT HaveLibHdfs3();
} // namespace io
} // namespace arrow
cpp/src/arrow/io/io-hdfs-test.cc
+122 -93
View File
@@ -24,6 +24,7 @@
#include <boost/filesystem.hpp> // NOLINT
#include "arrow/io/hdfs-internal.h"
#include "arrow/io/hdfs.h"
#include "arrow/status.h"
#include "arrow/test-util.h"
@@ -37,6 +38,7 @@ std::vector<uint8_t> RandomData(int64_t size) {
return buffer;
}
template <typename DRIVER>
class TestHdfsClient : public ::testing::Test {
public:
Status MakeScratchDir() {
@@ -71,15 +73,34 @@ class TestHdfsClient : public ::testing::Test {
return ss.str();
}
protected:
// Set up shared state between unit tests
static void SetUpTestCase() {
if (!ConnectLibHdfs().ok()) {
std::cout << "Loading libhdfs failed, skipping tests gracefully" << std::endl;
return;
void SetUp() {
LibHdfsShim* driver_shim;
client_ = nullptr;
scratch_dir_ =
boost::filesystem::unique_path("/tmp/arrow-hdfs/scratch-%%%%").native();
loaded_driver_ = false;
Status msg;
if (DRIVER::type == HdfsDriver::LIBHDFS) {
msg = ConnectLibHdfs(&driver_shim);
if (!msg.ok()) {
std::cout << "Loading libhdfs failed, skipping tests gracefully" << std::endl;
return;
}
} else {
msg = ConnectLibHdfs3(&driver_shim);
if (!msg.ok()) {
std::cout << "Loading libhdfs3 failed, skipping tests gracefully. "
<< msg.ToString() << std::endl;
return;
}
}
loaded_libhdfs_ = true;
loaded_driver_ = true;
const char* host = std::getenv("ARROW_HDFS_TEST_HOST");
const char* port = std::getenv("ARROW_HDFS_TEST_PORT");
@@ -94,151 +115,159 @@ class TestHdfsClient : public ::testing::Test {
ASSERT_OK(HdfsClient::Connect(&conf_, &client_));
}
static void TearDownTestCase() {
void TearDown() {
if (client_) {
EXPECT_OK(client_->Delete(scratch_dir_, true));
if (client_->Exists(scratch_dir_)) {
EXPECT_OK(client_->Delete(scratch_dir_, true));
}
EXPECT_OK(client_->Disconnect());
}
}
static bool loaded_libhdfs_;
HdfsConnectionConfig conf_;
bool loaded_driver_;
// Resources shared amongst unit tests
static HdfsConnectionConfig conf_;
static std::string scratch_dir_;
static std::shared_ptr<HdfsClient> client_;
std::string scratch_dir_;
std::shared_ptr<HdfsClient> client_;
};
bool TestHdfsClient::loaded_libhdfs_ = false;
HdfsConnectionConfig TestHdfsClient::conf_ = HdfsConnectionConfig();
std::string TestHdfsClient::scratch_dir_ =
boost::filesystem::unique_path("/tmp/arrow-hdfs/scratch-%%%%").native();
std::shared_ptr<HdfsClient> TestHdfsClient::client_ = nullptr;
#define SKIP_IF_NO_LIBHDFS() \
if (!loaded_libhdfs_) { \
std::cout << "No libhdfs, skipping" << std::endl; \
return; \
#define SKIP_IF_NO_DRIVER() \
if (!this->loaded_driver_) { \
std::cout << "Driver not loaded, skipping" << std::endl; \
return; \
}
TEST_F(TestHdfsClient, ConnectsAgain) {
SKIP_IF_NO_LIBHDFS();
struct JNIDriver {
static HdfsDriver type;
};
struct PivotalDriver {
static HdfsDriver type;
};
HdfsDriver JNIDriver::type = HdfsDriver::LIBHDFS;
HdfsDriver PivotalDriver::type = HdfsDriver::LIBHDFS3;
typedef ::testing::Types<JNIDriver, PivotalDriver> DriverTypes;
TYPED_TEST_CASE(TestHdfsClient, DriverTypes);
TYPED_TEST(TestHdfsClient, ConnectsAgain) {
SKIP_IF_NO_DRIVER();
std::shared_ptr<HdfsClient> client;
ASSERT_OK(HdfsClient::Connect(&conf_, &client));
ASSERT_OK(HdfsClient::Connect(&this->conf_, &client));
ASSERT_OK(client->Disconnect());
}
TEST_F(TestHdfsClient, CreateDirectory) {
SKIP_IF_NO_LIBHDFS();
TYPED_TEST(TestHdfsClient, CreateDirectory) {
SKIP_IF_NO_DRIVER();
std::string path = ScratchPath("create-directory");
std::string path = this->ScratchPath("create-directory");
if (client_->Exists(path)) { ASSERT_OK(client_->Delete(path, true)); }
if (this->client_->Exists(path)) { ASSERT_OK(this->client_->Delete(path, true)); }
ASSERT_OK(client_->CreateDirectory(path));
ASSERT_TRUE(client_->Exists(path));
EXPECT_OK(client_->Delete(path, true));
ASSERT_FALSE(client_->Exists(path));
ASSERT_OK(this->client_->CreateDirectory(path));
ASSERT_TRUE(this->client_->Exists(path));
EXPECT_OK(this->client_->Delete(path, true));
ASSERT_FALSE(this->client_->Exists(path));
}
TEST_F(TestHdfsClient, GetCapacityUsed) {
SKIP_IF_NO_LIBHDFS();
TYPED_TEST(TestHdfsClient, GetCapacityUsed) {
SKIP_IF_NO_DRIVER();
// Who knows what is actually in your DFS cluster, but expect it to have
// positive used bytes and capacity
int64_t nbytes = 0;
ASSERT_OK(client_->GetCapacity(&nbytes));
ASSERT_OK(this->client_->GetCapacity(&nbytes));
ASSERT_LT(0, nbytes);
ASSERT_OK(client_->GetUsed(&nbytes));
ASSERT_OK(this->client_->GetUsed(&nbytes));
ASSERT_LT(0, nbytes);
}
TEST_F(TestHdfsClient, GetPathInfo) {
SKIP_IF_NO_LIBHDFS();
TYPED_TEST(TestHdfsClient, GetPathInfo) {
SKIP_IF_NO_DRIVER();
HdfsPathInfo info;
ASSERT_OK(MakeScratchDir());
ASSERT_OK(this->MakeScratchDir());
// Directory info
ASSERT_OK(client_->GetPathInfo(scratch_dir_, &info));
ASSERT_OK(this->client_->GetPathInfo(this->scratch_dir_, &info));
ASSERT_EQ(ObjectType::DIRECTORY, info.kind);
ASSERT_EQ(HdfsAbsPath(scratch_dir_), info.name);
ASSERT_EQ(conf_.user, info.owner);
ASSERT_EQ(this->HdfsAbsPath(this->scratch_dir_), info.name);
ASSERT_EQ(this->conf_.user, info.owner);
// TODO(wesm): test group, other attrs
auto path = ScratchPath("test-file");
auto path = this->ScratchPath("test-file");
const int size = 100;
std::vector<uint8_t> buffer = RandomData(size);
ASSERT_OK(WriteDummyFile(path, buffer.data(), size));
ASSERT_OK(client_->GetPathInfo(path, &info));
ASSERT_OK(this->WriteDummyFile(path, buffer.data(), size));
ASSERT_OK(this->client_->GetPathInfo(path, &info));
ASSERT_EQ(ObjectType::FILE, info.kind);
ASSERT_EQ(HdfsAbsPath(path), info.name);
ASSERT_EQ(conf_.user, info.owner);
ASSERT_EQ(this->HdfsAbsPath(path), info.name);
ASSERT_EQ(this->conf_.user, info.owner);
ASSERT_EQ(size, info.size);
}
TEST_F(TestHdfsClient, AppendToFile) {
SKIP_IF_NO_LIBHDFS();
TYPED_TEST(TestHdfsClient, AppendToFile) {
SKIP_IF_NO_DRIVER();
ASSERT_OK(MakeScratchDir());
ASSERT_OK(this->MakeScratchDir());
auto path = ScratchPath("test-file");
auto path = this->ScratchPath("test-file");
const int size = 100;
std::vector<uint8_t> buffer = RandomData(size);
ASSERT_OK(WriteDummyFile(path, buffer.data(), size));
ASSERT_OK(this->WriteDummyFile(path, buffer.data(), size));
// now append
ASSERT_OK(WriteDummyFile(path, buffer.data(), size, true));
ASSERT_OK(this->WriteDummyFile(path, buffer.data(), size, true));
HdfsPathInfo info;
ASSERT_OK(client_->GetPathInfo(path, &info));
ASSERT_OK(this->client_->GetPathInfo(path, &info));
ASSERT_EQ(size * 2, info.size);
}
TEST_F(TestHdfsClient, ListDirectory) {
SKIP_IF_NO_LIBHDFS();
TYPED_TEST(TestHdfsClient, ListDirectory) {
SKIP_IF_NO_DRIVER();
const int size = 100;
std::vector<uint8_t> data = RandomData(size);
auto p1 = ScratchPath("test-file-1");
auto p2 = ScratchPath("test-file-2");
auto d1 = ScratchPath("test-dir-1");
auto p1 = this->ScratchPath("test-file-1");
auto p2 = this->ScratchPath("test-file-2");
auto d1 = this->ScratchPath("test-dir-1");
ASSERT_OK(MakeScratchDir());
ASSERT_OK(WriteDummyFile(p1, data.data(), size));
ASSERT_OK(WriteDummyFile(p2, data.data(), size / 2));
ASSERT_OK(client_->CreateDirectory(d1));
ASSERT_OK(this->MakeScratchDir());
ASSERT_OK(this->WriteDummyFile(p1, data.data(), size));
ASSERT_OK(this->WriteDummyFile(p2, data.data(), size / 2));
ASSERT_OK(this->client_->CreateDirectory(d1));
std::vector<HdfsPathInfo> listing;
ASSERT_OK(client_->ListDirectory(scratch_dir_, &listing));
ASSERT_OK(this->client_->ListDirectory(this->scratch_dir_, &listing));
// Do it again, appends!
ASSERT_OK(client_->ListDirectory(scratch_dir_, &listing));
ASSERT_OK(this->client_->ListDirectory(this->scratch_dir_, &listing));
ASSERT_EQ(6, static_cast<int>(listing.size()));
// Argh, well, shouldn't expect the listing to be in any particular order
for (size_t i = 0; i < listing.size(); ++i) {
const HdfsPathInfo& info = listing[i];
if (info.name == HdfsAbsPath(p1)) {
if (info.name == this->HdfsAbsPath(p1)) {
ASSERT_EQ(ObjectType::FILE, info.kind);
ASSERT_EQ(size, info.size);
} else if (info.name == HdfsAbsPath(p2)) {
} else if (info.name == this->HdfsAbsPath(p2)) {
ASSERT_EQ(ObjectType::FILE, info.kind);
ASSERT_EQ(size / 2, info.size);
} else if (info.name == HdfsAbsPath(d1)) {
} else if (info.name == this->HdfsAbsPath(d1)) {
ASSERT_EQ(ObjectType::DIRECTORY, info.kind);
} else {
FAIL() << "Unexpected path: " << info.name;
@@ -246,19 +275,19 @@ TEST_F(TestHdfsClient, ListDirectory) {
}
}
TEST_F(TestHdfsClient, ReadableMethods) {
SKIP_IF_NO_LIBHDFS();
TYPED_TEST(TestHdfsClient, ReadableMethods) {
SKIP_IF_NO_DRIVER();
ASSERT_OK(MakeScratchDir());
ASSERT_OK(this->MakeScratchDir());
auto path = ScratchPath("test-file");
auto path = this->ScratchPath("test-file");
const int size = 100;
std::vector<uint8_t> data = RandomData(size);
ASSERT_OK(WriteDummyFile(path, data.data(), size));
ASSERT_OK(this->WriteDummyFile(path, data.data(), size));
std::shared_ptr<HdfsReadableFile> file;
ASSERT_OK(client_->OpenReadable(path, &file));
ASSERT_OK(this->client_->OpenReadable(path, &file));
// Test GetSize -- move this into its own unit test if ever needed
int64_t file_size;
@@ -293,19 +322,19 @@ TEST_F(TestHdfsClient, ReadableMethods) {
ASSERT_EQ(60, position);
}
TEST_F(TestHdfsClient, LargeFile) {
SKIP_IF_NO_LIBHDFS();
TYPED_TEST(TestHdfsClient, LargeFile) {
SKIP_IF_NO_DRIVER();
ASSERT_OK(MakeScratchDir());
ASSERT_OK(this->MakeScratchDir());
auto path = ScratchPath("test-large-file");
auto path = this->ScratchPath("test-large-file");
const int size = 1000000;
std::vector<uint8_t> data = RandomData(size);
ASSERT_OK(WriteDummyFile(path, data.data(), size));
ASSERT_OK(this->WriteDummyFile(path, data.data(), size));
std::shared_ptr<HdfsReadableFile> file;
ASSERT_OK(client_->OpenReadable(path, &file));
ASSERT_OK(this->client_->OpenReadable(path, &file));
auto buffer = std::make_shared<PoolBuffer>();
ASSERT_OK(buffer->Resize(size));
@@ -317,7 +346,7 @@ TEST_F(TestHdfsClient, LargeFile) {
// explicit buffer size
std::shared_ptr<HdfsReadableFile> file2;
ASSERT_OK(client_->OpenReadable(path, 1 << 18, &file2));
ASSERT_OK(this->client_->OpenReadable(path, 1 << 18, &file2));
auto buffer2 = std::make_shared<PoolBuffer>();
ASSERT_OK(buffer2->Resize(size));
@@ -326,22 +355,22 @@ TEST_F(TestHdfsClient, LargeFile) {
ASSERT_EQ(size, bytes_read);
}
TEST_F(TestHdfsClient, RenameFile) {
SKIP_IF_NO_LIBHDFS();
TYPED_TEST(TestHdfsClient, RenameFile) {
SKIP_IF_NO_DRIVER();
ASSERT_OK(MakeScratchDir());
ASSERT_OK(this->MakeScratchDir());
auto src_path = ScratchPath("src-file");
auto dst_path = ScratchPath("dst-file");
auto src_path = this->ScratchPath("src-file");
auto dst_path = this->ScratchPath("dst-file");
const int size = 100;
std::vector<uint8_t> data = RandomData(size);
ASSERT_OK(WriteDummyFile(src_path, data.data(), size));
ASSERT_OK(this->WriteDummyFile(src_path, data.data(), size));
ASSERT_OK(client_->Rename(src_path, dst_path));
ASSERT_OK(this->client_->Rename(src_path, dst_path));
ASSERT_FALSE(client_->Exists(src_path));
ASSERT_TRUE(client_->Exists(dst_path));
ASSERT_FALSE(this->client_->Exists(src_path));
ASSERT_TRUE(this->client_->Exists(dst_path));
}
} // namespace io
cpp/src/arrow/io/libhdfs_shim.cc
-582
View File
@@ -1,582 +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.
// This shim interface to libhdfs (for runtime shared library loading) has been
// adapted from the SFrame project, released under the ASF-compatible 3-clause
// BSD license
//
// Using this required having the $JAVA_HOME and $HADOOP_HOME environment
// variables set, so that libjvm and libhdfs can be located easily
// Copyright (C) 2015 Dato, Inc.
// All rights reserved.
//
// This software may be modified and distributed under the terms
// of the BSD license. See the LICENSE file for details.
#ifdef HAS_HADOOP
#ifndef _WIN32
#include <dlfcn.h>
#else
#include <windows.h>
#include <winsock2.h>
// TODO(wesm): address when/if we add windows support
// #include <util/syserr_reporting.hpp>
#endif
extern "C" {
#include <hdfs.h>
}
#include <iostream>
#include <mutex>
#include <sstream>
#include <string>
#include <type_traits>
#include <vector>
#include <boost/filesystem.hpp> // NOLINT
#include "arrow/status.h"
#include "arrow/util/visibility.h"
namespace fs = boost::filesystem;
extern "C" {
#ifndef _WIN32
static void* libhdfs_handle = NULL;
static void* libjvm_handle = NULL;
#else
static HINSTANCE libhdfs_handle = NULL;
static HINSTANCE libjvm_handle = NULL;
#endif
/*
* All the shim pointers
*/
// NOTE(wesm): cpplint does not like use of short and other imprecise C types
static hdfsBuilder* (*ptr_hdfsNewBuilder)(void) = NULL;
static void (*ptr_hdfsBuilderSetNameNode)(hdfsBuilder* bld, const char* nn) = NULL;
static void (*ptr_hdfsBuilderSetNameNodePort)(hdfsBuilder* bld, tPort port) = NULL;
static void (*ptr_hdfsBuilderSetUserName)(hdfsBuilder* bld, const char* userName) = NULL;
static void (*ptr_hdfsBuilderSetKerbTicketCachePath)(
hdfsBuilder* bld, const char* kerbTicketCachePath) = NULL;
static hdfsFS (*ptr_hdfsBuilderConnect)(hdfsBuilder* bld) = NULL;
static int (*ptr_hdfsDisconnect)(hdfsFS fs) = NULL;
static hdfsFile (*ptr_hdfsOpenFile)(hdfsFS fs, const char* path, int flags,
int bufferSize, short replication, tSize blocksize) = NULL; // NOLINT
static int (*ptr_hdfsCloseFile)(hdfsFS fs, hdfsFile file) = NULL;
static int (*ptr_hdfsExists)(hdfsFS fs, const char* path) = NULL;
static int (*ptr_hdfsSeek)(hdfsFS fs, hdfsFile file, tOffset desiredPos) = NULL;
static tOffset (*ptr_hdfsTell)(hdfsFS fs, hdfsFile file) = NULL;
static tSize (*ptr_hdfsRead)(hdfsFS fs, hdfsFile file, void* buffer, tSize length) = NULL;
static tSize (*ptr_hdfsPread)(
hdfsFS fs, hdfsFile file, tOffset position, void* buffer, tSize length) = NULL;
static tSize (*ptr_hdfsWrite)(
hdfsFS fs, hdfsFile file, const void* buffer, tSize length) = NULL;
static int (*ptr_hdfsFlush)(hdfsFS fs, hdfsFile file) = NULL;
static int (*ptr_hdfsAvailable)(hdfsFS fs, hdfsFile file) = NULL;
static int (*ptr_hdfsCopy)(
hdfsFS srcFS, const char* src, hdfsFS dstFS, const char* dst) = NULL;
static int (*ptr_hdfsMove)(
hdfsFS srcFS, const char* src, hdfsFS dstFS, const char* dst) = NULL;
static int (*ptr_hdfsDelete)(hdfsFS fs, const char* path, int recursive) = NULL;
static int (*ptr_hdfsRename)(hdfsFS fs, const char* oldPath, const char* newPath) = NULL;
static char* (*ptr_hdfsGetWorkingDirectory)(
hdfsFS fs, char* buffer, size_t bufferSize) = NULL;
static int (*ptr_hdfsSetWorkingDirectory)(hdfsFS fs, const char* path) = NULL;
static int (*ptr_hdfsCreateDirectory)(hdfsFS fs, const char* path) = NULL;
static int (*ptr_hdfsSetReplication)(
hdfsFS fs, const char* path, int16_t replication) = NULL;
static hdfsFileInfo* (*ptr_hdfsListDirectory)(
hdfsFS fs, const char* path, int* numEntries) = NULL;
static hdfsFileInfo* (*ptr_hdfsGetPathInfo)(hdfsFS fs, const char* path) = NULL;
static void (*ptr_hdfsFreeFileInfo)(hdfsFileInfo* hdfsFileInfo, int numEntries) = NULL;
static char*** (*ptr_hdfsGetHosts)(
hdfsFS fs, const char* path, tOffset start, tOffset length) = NULL;
static void (*ptr_hdfsFreeHosts)(char*** blockHosts) = NULL;
static tOffset (*ptr_hdfsGetDefaultBlockSize)(hdfsFS fs) = NULL;
static tOffset (*ptr_hdfsGetCapacity)(hdfsFS fs) = NULL;
static tOffset (*ptr_hdfsGetUsed)(hdfsFS fs) = NULL;
static int (*ptr_hdfsChown)(
hdfsFS fs, const char* path, const char* owner, const char* group) = NULL;
static int (*ptr_hdfsChmod)(hdfsFS fs, const char* path, short mode) = NULL; // NOLINT
static int (*ptr_hdfsUtime)(hdfsFS fs, const char* path, tTime mtime, tTime atime) = NULL;
// Helper functions for dlopens
static std::vector<fs::path> get_potential_libjvm_paths();
static std::vector<fs::path> get_potential_libhdfs_paths();
static arrow::Status try_dlopen(std::vector<fs::path> potential_paths, const char* name,
#ifndef _WIN32
void*& out_handle);
#else
HINSTANCE& out_handle);
#endif
#define GET_SYMBOL(SYMBOL_NAME) \
if (!ptr_##SYMBOL_NAME) { \
*reinterpret_cast<void**>(&ptr_##SYMBOL_NAME) = get_symbol("" #SYMBOL_NAME); \
}
static void* get_symbol(const char* symbol) {
if (libhdfs_handle == NULL) return NULL;
#ifndef _WIN32
return dlsym(libhdfs_handle, symbol);
#else
void* ret = reinterpret_cast<void*>(GetProcAddress(libhdfs_handle, symbol));
if (ret == NULL) {
// logstream(LOG_INFO) << "GetProcAddress error: "
// << get_last_err_str(GetLastError()) << std::endl;
}
return ret;
#endif
}
hdfsBuilder* hdfsNewBuilder(void) {
return ptr_hdfsNewBuilder();
}
void hdfsBuilderSetNameNode(hdfsBuilder* bld, const char* nn) {
ptr_hdfsBuilderSetNameNode(bld, nn);
}
void hdfsBuilderSetNameNodePort(hdfsBuilder* bld, tPort port) {
ptr_hdfsBuilderSetNameNodePort(bld, port);
}
void hdfsBuilderSetUserName(hdfsBuilder* bld, const char* userName) {
ptr_hdfsBuilderSetUserName(bld, userName);
}
void hdfsBuilderSetKerbTicketCachePath(
hdfsBuilder* bld, const char* kerbTicketCachePath) {
ptr_hdfsBuilderSetKerbTicketCachePath(bld, kerbTicketCachePath);
}
hdfsFS hdfsBuilderConnect(hdfsBuilder* bld) {
return ptr_hdfsBuilderConnect(bld);
}
int hdfsDisconnect(hdfsFS fs) {
return ptr_hdfsDisconnect(fs);
}
hdfsFile hdfsOpenFile(hdfsFS fs, const char* path, int flags, int bufferSize,
short replication, tSize blocksize) { // NOLINT
return ptr_hdfsOpenFile(fs, path, flags, bufferSize, replication, blocksize);
}
int hdfsCloseFile(hdfsFS fs, hdfsFile file) {
return ptr_hdfsCloseFile(fs, file);
}
int hdfsExists(hdfsFS fs, const char* path) {
return ptr_hdfsExists(fs, path);
}
int hdfsSeek(hdfsFS fs, hdfsFile file, tOffset desiredPos) {
return ptr_hdfsSeek(fs, file, desiredPos);
}
tOffset hdfsTell(hdfsFS fs, hdfsFile file) {
return ptr_hdfsTell(fs, file);
}
tSize hdfsRead(hdfsFS fs, hdfsFile file, void* buffer, tSize length) {
return ptr_hdfsRead(fs, file, buffer, length);
}
tSize hdfsPread(hdfsFS fs, hdfsFile file, tOffset position, void* buffer, tSize length) {
return ptr_hdfsPread(fs, file, position, buffer, length);
}
tSize hdfsWrite(hdfsFS fs, hdfsFile file, const void* buffer, tSize length) {
return ptr_hdfsWrite(fs, file, buffer, length);
}
int hdfsFlush(hdfsFS fs, hdfsFile file) {
return ptr_hdfsFlush(fs, file);
}
int hdfsAvailable(hdfsFS fs, hdfsFile file) {
GET_SYMBOL(hdfsAvailable);
if (ptr_hdfsAvailable)
return ptr_hdfsAvailable(fs, file);
else
return 0;
}
int hdfsCopy(hdfsFS srcFS, const char* src, hdfsFS dstFS, const char* dst) {
GET_SYMBOL(hdfsCopy);
if (ptr_hdfsCopy)
return ptr_hdfsCopy(srcFS, src, dstFS, dst);
else
return 0;
}
int hdfsMove(hdfsFS srcFS, const char* src, hdfsFS dstFS, const char* dst) {
GET_SYMBOL(hdfsMove);
if (ptr_hdfsMove)
return ptr_hdfsMove(srcFS, src, dstFS, dst);
else
return 0;
}
int hdfsDelete(hdfsFS fs, const char* path, int recursive) {
return ptr_hdfsDelete(fs, path, recursive);
}
int hdfsRename(hdfsFS fs, const char* oldPath, const char* newPath) {
GET_SYMBOL(hdfsRename);
if (ptr_hdfsRename)
return ptr_hdfsRename(fs, oldPath, newPath);
else
return 0;
}
char* hdfsGetWorkingDirectory(hdfsFS fs, char* buffer, size_t bufferSize) {
GET_SYMBOL(hdfsGetWorkingDirectory);
if (ptr_hdfsGetWorkingDirectory) {
return ptr_hdfsGetWorkingDirectory(fs, buffer, bufferSize);
} else {
return NULL;
}
}
int hdfsSetWorkingDirectory(hdfsFS fs, const char* path) {
GET_SYMBOL(hdfsSetWorkingDirectory);
if (ptr_hdfsSetWorkingDirectory) {
return ptr_hdfsSetWorkingDirectory(fs, path);
} else {
return 0;
}
}
int hdfsCreateDirectory(hdfsFS fs, const char* path) {
return ptr_hdfsCreateDirectory(fs, path);
}
int hdfsSetReplication(hdfsFS fs, const char* path, int16_t replication) {
GET_SYMBOL(hdfsSetReplication);
if (ptr_hdfsSetReplication) {
return ptr_hdfsSetReplication(fs, path, replication);
} else {
return 0;
}
}
hdfsFileInfo* hdfsListDirectory(hdfsFS fs, const char* path, int* numEntries) {
return ptr_hdfsListDirectory(fs, path, numEntries);
}
hdfsFileInfo* hdfsGetPathInfo(hdfsFS fs, const char* path) {
return ptr_hdfsGetPathInfo(fs, path);
}
void hdfsFreeFileInfo(hdfsFileInfo* hdfsFileInfo, int numEntries) {
ptr_hdfsFreeFileInfo(hdfsFileInfo, numEntries);
}
char*** hdfsGetHosts(hdfsFS fs, const char* path, tOffset start, tOffset length) {
GET_SYMBOL(hdfsGetHosts);
if (ptr_hdfsGetHosts) {
return ptr_hdfsGetHosts(fs, path, start, length);
} else {
return NULL;
}
}
void hdfsFreeHosts(char*** blockHosts) {
GET_SYMBOL(hdfsFreeHosts);
if (ptr_hdfsFreeHosts) { ptr_hdfsFreeHosts(blockHosts); }
}
tOffset hdfsGetDefaultBlockSize(hdfsFS fs) {
GET_SYMBOL(hdfsGetDefaultBlockSize);
if (ptr_hdfsGetDefaultBlockSize) {
return ptr_hdfsGetDefaultBlockSize(fs);
} else {
return 0;
}
}
tOffset hdfsGetCapacity(hdfsFS fs) {
return ptr_hdfsGetCapacity(fs);
}
tOffset hdfsGetUsed(hdfsFS fs) {
return ptr_hdfsGetUsed(fs);
}
int hdfsChown(hdfsFS fs, const char* path, const char* owner, const char* group) {
GET_SYMBOL(hdfsChown);
if (ptr_hdfsChown) {
return ptr_hdfsChown(fs, path, owner, group);
} else {
return 0;
}
}
int hdfsChmod(hdfsFS fs, const char* path, short mode) { // NOLINT
GET_SYMBOL(hdfsChmod);
if (ptr_hdfsChmod) {
return ptr_hdfsChmod(fs, path, mode);
} else {
return 0;
}
}
int hdfsUtime(hdfsFS fs, const char* path, tTime mtime, tTime atime) {
GET_SYMBOL(hdfsUtime);
if (ptr_hdfsUtime) {
return ptr_hdfsUtime(fs, path, mtime, atime);
} else {
return 0;
}
}
static std::vector<fs::path> get_potential_libhdfs_paths() {
std::vector<fs::path> libhdfs_potential_paths;
std::string file_name;
// OS-specific file name
#ifdef __WIN32
file_name = "hdfs.dll";
#elif __APPLE__
file_name = "libhdfs.dylib";
#else
file_name = "libhdfs.so";
#endif
// Common paths
std::vector<fs::path> search_paths = {fs::path(""), fs::path(".")};
// Path from environment variable
const char* hadoop_home = std::getenv("HADOOP_HOME");
if (hadoop_home != nullptr) {
auto path = fs::path(hadoop_home) / "lib/native";
search_paths.push_back(path);
}
const char* libhdfs_dir = std::getenv("ARROW_LIBHDFS_DIR");
if (libhdfs_dir != nullptr) { search_paths.push_back(fs::path(libhdfs_dir)); }
// All paths with file name
for (auto& path : search_paths) {
libhdfs_potential_paths.push_back(path / file_name);
}
return libhdfs_potential_paths;
}
static std::vector<fs::path> get_potential_libjvm_paths() {
std::vector<fs::path> libjvm_potential_paths;
std::vector<fs::path> search_prefixes;
std::vector<fs::path> search_suffixes;
std::string file_name;
// From heuristics
#ifdef __WIN32
search_prefixes = {""};
search_suffixes = {"/jre/bin/server", "/bin/server"};
file_name = "jvm.dll";
#elif __APPLE__
search_prefixes = {""};
search_suffixes = {"", "/jre/lib/server"};
file_name = "libjvm.dylib";
// SFrame uses /usr/libexec/java_home to find JAVA_HOME; for now we are
// expecting users to set an environment variable
#else
search_prefixes = {
"/usr/lib/jvm/default-java", // ubuntu / debian distros
"/usr/lib/jvm/java", // rhel6
"/usr/lib/jvm", // centos6
"/usr/lib64/jvm", // opensuse 13
"/usr/local/lib/jvm/default-java", // alt ubuntu / debian distros
"/usr/local/lib/jvm/java", // alt rhel6
"/usr/local/lib/jvm", // alt centos6
"/usr/local/lib64/jvm", // alt opensuse 13
"/usr/local/lib/jvm/java-7-openjdk-amd64", // alt ubuntu / debian distros
"/usr/lib/jvm/java-7-openjdk-amd64", // alt ubuntu / debian distros
"/usr/local/lib/jvm/java-6-openjdk-amd64", // alt ubuntu / debian distros
"/usr/lib/jvm/java-6-openjdk-amd64", // alt ubuntu / debian distros
"/usr/lib/jvm/java-7-oracle", // alt ubuntu
"/usr/lib/jvm/java-8-oracle", // alt ubuntu
"/usr/lib/jvm/java-6-oracle", // alt ubuntu
"/usr/local/lib/jvm/java-7-oracle", // alt ubuntu
"/usr/local/lib/jvm/java-8-oracle", // alt ubuntu
"/usr/local/lib/jvm/java-6-oracle", // alt ubuntu
"/usr/lib/jvm/default", // alt centos
"/usr/java/latest", // alt centos
};
search_suffixes = {"/jre/lib/amd64/server"};
file_name = "libjvm.so";
#endif
// From direct environment variable
char* env_value = NULL;
if ((env_value = getenv("JAVA_HOME")) != NULL) {
// logstream(LOG_INFO) << "Found environment variable " << env_name << ": " <<
// env_value << std::endl;
search_prefixes.insert(search_prefixes.begin(), env_value);
}
// Generate cross product between search_prefixes, search_suffixes, and file_name
for (auto& prefix : search_prefixes) {
for (auto& suffix : search_suffixes) {
auto path = (fs::path(prefix) / fs::path(suffix) / fs::path(file_name));
libjvm_potential_paths.push_back(path);
}
}
return libjvm_potential_paths;
}
#ifndef _WIN32
static arrow::Status try_dlopen(
std::vector<fs::path> potential_paths, const char* name, void*& out_handle) {
std::vector<std::string> error_messages;
for (auto& i : potential_paths) {
i.make_preferred();
// logstream(LOG_INFO) << "Trying " << i.string().c_str() << std::endl;
out_handle = dlopen(i.native().c_str(), RTLD_NOW | RTLD_LOCAL);
if (out_handle != NULL) {
// logstream(LOG_INFO) << "Success!" << std::endl;
break;
} else {
const char* err_msg = dlerror();
if (err_msg != NULL) {
error_messages.push_back(std::string(err_msg));
} else {
error_messages.push_back(std::string(" returned NULL"));
}
}
}
if (out_handle == NULL) {
std::stringstream ss;
ss << "Unable to load " << name;
return arrow::Status::IOError(ss.str());
}
return arrow::Status::OK();
}
#else
static arrow::Status try_dlopen(
std::vector<fs::path> potential_paths, const char* name, HINSTANCE& out_handle) {
std::vector<std::string> error_messages;
for (auto& i : potential_paths) {
i.make_preferred();
// logstream(LOG_INFO) << "Trying " << i.string().c_str() << std::endl;
out_handle = LoadLibrary(i.string().c_str());
if (out_handle != NULL) {
// logstream(LOG_INFO) << "Success!" << std::endl;
break;
} else {
// error_messages.push_back(get_last_err_str(GetLastError()));
}
}
if (out_handle == NULL) {
std::stringstream ss;
ss << "Unable to load " << name;
return arrow::Status::IOError(ss.str());
}
return arrow::Status::OK();
}
#endif // _WIN32
} // extern "C"
#define GET_SYMBOL_REQUIRED(SYMBOL_NAME) \
do { \
if (!ptr_##SYMBOL_NAME) { \
*reinterpret_cast<void**>(&ptr_##SYMBOL_NAME) = get_symbol("" #SYMBOL_NAME); \
} \
if (!ptr_##SYMBOL_NAME) \
return Status::IOError("Getting symbol " #SYMBOL_NAME "failed"); \
} while (0)
namespace arrow {
namespace io {
Status ARROW_EXPORT ConnectLibHdfs() {
static std::mutex lock;
std::lock_guard<std::mutex> guard(lock);
static bool shim_attempted = false;
if (!shim_attempted) {
shim_attempted = true;
std::vector<fs::path> libjvm_potential_paths = get_potential_libjvm_paths();
RETURN_NOT_OK(try_dlopen(libjvm_potential_paths, "libjvm", libjvm_handle));
std::vector<fs::path> libhdfs_potential_paths = get_potential_libhdfs_paths();
RETURN_NOT_OK(try_dlopen(libhdfs_potential_paths, "libhdfs", libhdfs_handle));
} else if (libhdfs_handle == nullptr) {
return Status::IOError("Prior attempt to load libhdfs failed");
}
GET_SYMBOL_REQUIRED(hdfsNewBuilder);
GET_SYMBOL_REQUIRED(hdfsBuilderSetNameNode);
GET_SYMBOL_REQUIRED(hdfsBuilderSetNameNodePort);
GET_SYMBOL_REQUIRED(hdfsBuilderSetUserName);
GET_SYMBOL_REQUIRED(hdfsBuilderSetKerbTicketCachePath);
GET_SYMBOL_REQUIRED(hdfsBuilderConnect);
GET_SYMBOL_REQUIRED(hdfsCreateDirectory);
GET_SYMBOL_REQUIRED(hdfsDelete);
GET_SYMBOL_REQUIRED(hdfsDisconnect);
GET_SYMBOL_REQUIRED(hdfsExists);
GET_SYMBOL_REQUIRED(hdfsFreeFileInfo);
GET_SYMBOL_REQUIRED(hdfsGetCapacity);
GET_SYMBOL_REQUIRED(hdfsGetUsed);
GET_SYMBOL_REQUIRED(hdfsGetPathInfo);
GET_SYMBOL_REQUIRED(hdfsListDirectory);
// File methods
GET_SYMBOL_REQUIRED(hdfsCloseFile);
GET_SYMBOL_REQUIRED(hdfsFlush);
GET_SYMBOL_REQUIRED(hdfsOpenFile);
GET_SYMBOL_REQUIRED(hdfsRead);
GET_SYMBOL_REQUIRED(hdfsPread);
GET_SYMBOL_REQUIRED(hdfsSeek);
GET_SYMBOL_REQUIRED(hdfsTell);
GET_SYMBOL_REQUIRED(hdfsWrite);
return Status::OK();
}
} // namespace io
} // namespace arrow
#endif // HAS_HADOOP
python/.gitignore
+1
View File
@@ -16,6 +16,7 @@ Testing/
*.c
*.cpp
pyarrow/version.py
pyarrow/table_api.h
# Python files
# setup.py working directory
python/pyarrow/includes/libarrow_io.pxd
+7 -1
View File
@@ -87,13 +87,19 @@ cdef extern from "arrow/io/file.h" namespace "arrow::io" nogil:
cdef extern from "arrow/io/hdfs.h" namespace "arrow::io" nogil:
CStatus ConnectLibHdfs()
CStatus HaveLibHdfs()
CStatus HaveLibHdfs3()
enum HdfsDriver" arrow::io::HdfsDriver":
HdfsDriver_LIBHDFS" arrow::io::HdfsDriver::LIBHDFS"
HdfsDriver_LIBHDFS3" arrow::io::HdfsDriver::LIBHDFS3"
cdef cppclass HdfsConnectionConfig:
c_string host
int port
c_string user
c_string kerb_ticket
HdfsDriver driver
cdef cppclass HdfsPathInfo:
ObjectType kind;
python/pyarrow/io.pyx
+31 -14
View File
@@ -35,6 +35,7 @@ cimport cpython as cp
import re
import sys
import threading
import time
cdef class NativeFile:
@@ -269,7 +270,15 @@ except ImportError:
def have_libhdfs():
try:
check_status(ConnectLibHdfs())
check_status(HaveLibHdfs())
return True
except:
return False
def have_libhdfs3():
try:
check_status(HaveLibHdfs3())
return True
except:
return False
@@ -313,7 +322,8 @@ cdef class HdfsClient:
raise IOError('HDFS client is closed')
@classmethod
def connect(cls, host="default", port=0, user=None, kerb_ticket=None):
def connect(cls, host="default", port=0, user=None, kerb_ticket=None,
driver='libhdfs'):
"""
Connect to an HDFS cluster. All parameters are optional and should
only be set if the defaults need to be overridden.
@@ -328,6 +338,9 @@ cdef class HdfsClient:
port : NameNode's port. Set to 0 for default or logical (HA) nodes.
user : Username when connecting to HDFS; None implies login user.
kerb_ticket : Path to Kerberos ticket cache.
driver : {'libhdfs', 'libhdfs3'}, default 'libhdfs'
Connect using libhdfs (JNI-based) or libhdfs3 (3rd-party C++
library from Pivotal Labs)
Notes
-----
@@ -350,6 +363,13 @@ cdef class HdfsClient:
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, &out.client))
out.is_open = True
@@ -541,6 +561,12 @@ cdef class HdfsClient:
if not buf:
break
if writer_thread.is_alive():
while write_queue.full():
time.sleep(0.01)
else:
break
write_queue.put_nowait(buf)
finally:
done = True
@@ -609,22 +635,13 @@ cdef class HdfsFile(NativeFile):
cdef int64_t total_bytes = 0
cdef int rpc_chunksize = min(self.buffer_size, nbytes)
try:
with nogil:
while total_bytes < nbytes:
check_status(self.rd_file.get()
.Read(rpc_chunksize, &bytes_read,
buf + total_bytes))
check_status(self.rd_file.get()
.Read(nbytes, &bytes_read, buf))
total_bytes += bytes_read
# EOF
if bytes_read == 0:
break
result = cp.PyBytes_FromStringAndSize(<const char*>buf,
total_bytes)
bytes_read)
finally:
free(buf)
python/pyarrow/tests/test_hdfs.py
+110 -105
View File
@@ -19,6 +19,7 @@ from io import BytesIO
from os.path import join as pjoin
import os
import random
import unittest
import pytest
@@ -28,7 +29,7 @@ import pyarrow.io as io
# HDFS tests
def hdfs_test_client():
def hdfs_test_client(driver='libhdfs'):
host = os.environ.get('ARROW_HDFS_TEST_HOST', 'localhost')
user = os.environ['ARROW_HDFS_TEST_USER']
try:
@@ -37,115 +38,119 @@ def hdfs_test_client():
raise ValueError('Env variable ARROW_HDFS_TEST_PORT was not '
'an integer')
return io.HdfsClient.connect(host, port, user)
return io.HdfsClient.connect(host, port, user, driver=driver)
libhdfs = pytest.mark.skipif(not io.have_libhdfs(),
reason='No libhdfs available on system')
class HdfsTestCases(object):
def _make_test_file(self, hdfs, test_name, test_path, test_data):
base_path = pjoin(self.tmp_path, test_name)
hdfs.mkdir(base_path)
full_path = pjoin(base_path, test_path)
with hdfs.open(full_path, 'wb') as f:
f.write(test_data)
return full_path
@classmethod
def setUpClass(cls):
cls.check_driver()
cls.hdfs = hdfs_test_client(cls.DRIVER)
cls.tmp_path = '/tmp/pyarrow-test-{0}'.format(random.randint(0, 1000))
cls.hdfs.mkdir(cls.tmp_path)
@classmethod
def tearDownClass(cls):
cls.hdfs.delete(cls.tmp_path, recursive=True)
cls.hdfs.close()
def test_hdfs_close(self):
client = hdfs_test_client()
assert client.is_open
client.close()
assert not client.is_open
with pytest.raises(Exception):
client.ls('/')
def test_hdfs_mkdir(self):
path = pjoin(self.tmp_path, 'test-dir/test-dir')
parent_path = pjoin(self.tmp_path, 'test-dir')
self.hdfs.mkdir(path)
assert self.hdfs.exists(path)
self.hdfs.delete(parent_path, recursive=True)
assert not self.hdfs.exists(path)
def test_hdfs_ls(self):
base_path = pjoin(self.tmp_path, 'ls-test')
self.hdfs.mkdir(base_path)
dir_path = pjoin(base_path, 'a-dir')
f1_path = pjoin(base_path, 'a-file-1')
self.hdfs.mkdir(dir_path)
f = self.hdfs.open(f1_path, 'wb')
f.write('a' * 10)
contents = sorted(self.hdfs.ls(base_path, False))
assert contents == [dir_path, f1_path]
def test_hdfs_download_upload(self):
base_path = pjoin(self.tmp_path, 'upload-test')
data = b'foobarbaz'
buf = BytesIO(data)
buf.seek(0)
self.hdfs.upload(base_path, buf)
out_buf = BytesIO()
self.hdfs.download(base_path, out_buf)
out_buf.seek(0)
assert out_buf.getvalue() == data
def test_hdfs_file_context_manager(self):
path = pjoin(self.tmp_path, 'ctx-manager')
data = b'foo'
with self.hdfs.open(path, 'wb') as f:
f.write(data)
with self.hdfs.open(path, 'rb') as f:
assert f.size() == 3
result = f.read(10)
assert result == data
HDFS_TMP_PATH = '/tmp/pyarrow-test-{0}'.format(random.randint(0, 1000))
class TestLibHdfs(HdfsTestCases, unittest.TestCase):
DRIVER = 'libhdfs'
@classmethod
def check_driver(cls):
if not io.have_libhdfs():
pytest.skip('No libhdfs available on system')
def test_hdfs_orphaned_file(self):
hdfs = hdfs_test_client()
file_path = self._make_test_file(hdfs, 'orphaned_file_test', 'fname',
'foobarbaz')
f = hdfs.open(file_path)
hdfs = None
f = None # noqa
@pytest.fixture(scope='session')
def hdfs(request):
fixture = hdfs_test_client()
class TestLibHdfs3(HdfsTestCases, unittest.TestCase):
def teardown():
fixture.delete(HDFS_TMP_PATH, recursive=True)
fixture.close()
request.addfinalizer(teardown)
return fixture
DRIVER = 'libhdfs3'
@libhdfs
def test_hdfs_close():
client = hdfs_test_client()
assert client.is_open
client.close()
assert not client.is_open
with pytest.raises(Exception):
client.ls('/')
@libhdfs
def test_hdfs_mkdir(hdfs):
path = pjoin(HDFS_TMP_PATH, 'test-dir/test-dir')
parent_path = pjoin(HDFS_TMP_PATH, 'test-dir')
hdfs.mkdir(path)
assert hdfs.exists(path)
hdfs.delete(parent_path, recursive=True)
assert not hdfs.exists(path)
@libhdfs
def test_hdfs_ls(hdfs):
base_path = pjoin(HDFS_TMP_PATH, 'ls-test')
hdfs.mkdir(base_path)
dir_path = pjoin(base_path, 'a-dir')
f1_path = pjoin(base_path, 'a-file-1')
hdfs.mkdir(dir_path)
f = hdfs.open(f1_path, 'wb')
f.write('a' * 10)
contents = sorted(hdfs.ls(base_path, False))
assert contents == [dir_path, f1_path]
def _make_test_file(hdfs, test_name, test_path, test_data):
base_path = pjoin(HDFS_TMP_PATH, test_name)
hdfs.mkdir(base_path)
full_path = pjoin(base_path, test_path)
f = hdfs.open(full_path, 'wb')
f.write(test_data)
return full_path
@libhdfs
def test_hdfs_orphaned_file():
hdfs = hdfs_test_client()
file_path = _make_test_file(hdfs, 'orphaned_file_test', 'fname',
'foobarbaz')
f = hdfs.open(file_path)
hdfs = None
f = None # noqa
@libhdfs
def test_hdfs_download_upload(hdfs):
base_path = pjoin(HDFS_TMP_PATH, 'upload-test')
data = b'foobarbaz'
buf = BytesIO(data)
buf.seek(0)
hdfs.upload(base_path, buf)
out_buf = BytesIO()
hdfs.download(base_path, out_buf)
out_buf.seek(0)
assert out_buf.getvalue() == data
@libhdfs
def test_hdfs_file_context_manager(hdfs):
path = pjoin(HDFS_TMP_PATH, 'ctx-manager')
data = b'foo'
with hdfs.open(path, 'wb') as f:
f.write(data)
with hdfs.open(path, 'rb') as f:
assert f.size() == 3
result = f.read(10)
assert result == data
@classmethod
def check_driver(cls):
if not io.have_libhdfs3():
pytest.skip('No libhdfs3 available on system')