You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by fo...@apache.org on 2023/07/26 07:27:52 UTC
[iceberg] branch master updated: Python: Add pyarrow hdfs support (#7997)
This is an automated email from the ASF dual-hosted git repository.
fokko pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/master by this push:
new d28f899c80 Python: Add pyarrow hdfs support (#7997)
d28f899c80 is described below
commit d28f899c802cb3b7b6bd4d6c44f6adf3a20a1bd9
Author: Luigi Cerone <lu...@gmail.com>
AuthorDate: Wed Jul 26 09:27:45 2023 +0200
Python: Add pyarrow hdfs support (#7997)
* Add pyarrow hdfs support
* Update PR review
---
python/mkdocs/docs/configuration.md | 9 +++++++++
python/pyiceberg/io/__init__.py | 4 ++++
python/pyiceberg/io/pyarrow.py | 13 +++++++++++++
python/tests/io/test_pyarrow.py | 15 +++++++++++++++
4 files changed, 41 insertions(+)
diff --git a/python/mkdocs/docs/configuration.md b/python/mkdocs/docs/configuration.md
index f6b24745c1..0510e4ff66 100644
--- a/python/mkdocs/docs/configuration.md
+++ b/python/mkdocs/docs/configuration.md
@@ -69,6 +69,15 @@ For the FileIO there are several configuration options available:
| s3.region | us-west-2 | Sets the region of the bucket |
| s3.proxy-uri | http://my.proxy.com:8080 | Configure the proxy server to be used by the FileIO. |
+### HDFS
+
+| Key | Example | Description |
+| -------------------- | ------------------- | ------------------------------------------------ |
+| hdfs.host | https://10.0.19.25/ | Configure the HDFS host to connect to |
+| hdfs.port | 9000 | Configure the HDFS port to connect to. |
+| hdfs.user | user | Configure the HDFS username used for connection. |
+| hdfs.kerberos_ticket | kerberos_ticket | Configure the path to the Kerberos ticket cache. |
+
### Azure Data lake
| Key | Example | Description |
diff --git a/python/pyiceberg/io/__init__.py b/python/pyiceberg/io/__init__.py
index c477c1ac4c..b611a91d24 100644
--- a/python/pyiceberg/io/__init__.py
+++ b/python/pyiceberg/io/__init__.py
@@ -51,6 +51,10 @@ S3_SECRET_ACCESS_KEY = "s3.secret-access-key"
S3_SESSION_TOKEN = "s3.session-token"
S3_REGION = "s3.region"
S3_PROXY_URI = "s3.proxy-uri"
+HDFS_HOST = "hdfs.host"
+HDFS_PORT = "hdfs.port"
+HDFS_USER = "hdfs.user"
+HDFS_KERB_TICKET = "hdfs.kerberos_ticket"
@runtime_checkable
diff --git a/python/pyiceberg/io/pyarrow.py b/python/pyiceberg/io/pyarrow.py
index c501278492..4255072003 100644
--- a/python/pyiceberg/io/pyarrow.py
+++ b/python/pyiceberg/io/pyarrow.py
@@ -57,6 +57,7 @@ from pyarrow.fs import (
FileSystem,
FileType,
FSSpecHandler,
+ HadoopFileSystem,
LocalFileSystem,
PyFileSystem,
S3FileSystem,
@@ -77,6 +78,10 @@ from pyiceberg.expressions.visitors import (
)
from pyiceberg.expressions.visitors import visit as boolean_expression_visit
from pyiceberg.io import (
+ HDFS_HOST,
+ HDFS_KERB_TICKET,
+ HDFS_PORT,
+ HDFS_USER,
S3_ACCESS_KEY_ID,
S3_ENDPOINT,
S3_PROXY_URI,
@@ -300,6 +305,14 @@ class PyArrowFileIO(FileIO):
client_kwargs["proxy_options"] = proxy_uri
return S3FileSystem(**client_kwargs)
+ elif scheme == "hdfs":
+ client_kwargs = {
+ "host": self.properties.get(HDFS_HOST),
+ "port": self.properties.get(HDFS_PORT),
+ "user": self.properties.get(HDFS_USER),
+ "kerb_ticket": self.properties.get(HDFS_KERB_TICKET),
+ }
+ return HadoopFileSystem(**client_kwargs)
elif scheme == "file":
return LocalFileSystem()
else:
diff --git a/python/tests/io/test_pyarrow.py b/python/tests/io/test_pyarrow.py
index 7a07532c66..01a8f35b1d 100644
--- a/python/tests/io/test_pyarrow.py
+++ b/python/tests/io/test_pyarrow.py
@@ -302,6 +302,21 @@ def test_deleting_s3_file_not_found() -> None:
assert "Cannot delete file, does not exist:" in str(exc_info.value)
+def test_deleting_hdfs_file_not_found() -> None:
+ """Test that a PyArrowFile raises a PermissionError when the pyarrow error includes 'No such file or directory'"""
+
+ hdfs_mock = MagicMock()
+ hdfs_mock.delete_file.side_effect = OSError("Path does not exist")
+
+ with patch.object(PyArrowFileIO, "_get_fs") as submocked:
+ submocked.return_value = hdfs_mock
+
+ with pytest.raises(FileNotFoundError) as exc_info:
+ PyArrowFileIO().delete("hdfs://foo/bar.txt")
+
+ assert "Cannot delete file, does not exist:" in str(exc_info.value)
+
+
def test_schema_to_pyarrow_schema(table_schema_nested: Schema) -> None:
actual = schema_to_pyarrow(table_schema_nested)
expected = """foo: string