Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Call globStatus directly via PY4J in hdfs_glob to avoid calling hadoop command #10599

Merged
merged 5 commits into from
Mar 19, 2024
Merged

Call globStatus directly via PY4J in hdfs_glob to avoid calling hadoop command #10599

Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
f932dcb
Call globStatus directly via PY4J instead of 'hadoop fs'
yinqingh Mar 15, 2024
07e0b3f
Use pathlib.Path to handle local path only
yinqingh Mar 16, 2024
825bb59
Use os.path.join to concatenate hdfs path and pattern
yinqingh Mar 16, 2024
428cd7a
Merge branch 'branch-24.04' into fix-hdfs-glob
yinqingh Mar 18, 2024
4d80814
Merge branch 'branch-24.04' into fix-hdfs-glob
yinqingh Mar 18, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 11 additions & 10 deletions integration_tests/src/main/python/parquet_testing_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,18 +81,19 @@ def hdfs_glob(path, pattern):
:type path: pathlib.Path
:return: generator of matched files
"""
from spark_init_internal import get_spark_i_know_what_i_am_doing
path_str = path.as_posix()
full_pattern = path_str + '/' + pattern
cmd = ['hadoop', 'fs', '-ls', '-C', full_pattern]

process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True)
jlowe marked this conversation as resolved.
Show resolved Hide resolved
stdout, stderr = process.communicate()
if process.returncode != 0:
raise AssertionError(f'Failed to list files from {path_str}. Error: {stderr}')

paths = stdout.strip().split('\n')

for p in paths:
sc = get_spark_i_know_what_i_am_doing().sparkContext
config = sc._jsc.hadoopConfiguration()
fs_path = sc._jvm.org.apache.hadoop.fs.Path(full_pattern)
fs = sc._jvm.org.apache.hadoop.fs.FileSystem.get(config)
jlowe marked this conversation as resolved.
Show resolved Hide resolved
statuses = fs.globStatus(fs_path)
for status in statuses:
# status.getPath().toString() return string like "hdfs://hostname:8020/src/test/resources/parquet-testing/data/single_nan.parquet"
# but pathlib.Path will remove the first "/" and convert it to "hdfs:/hostname:8020/src/test/resources/parquet-testing/data/single_nan.parquet" and then this path becomes illegal.
# so we need to process the path like this.
p = f'hdfs:{status.getPath().toUri().getPath()}'
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

you can get the scheme "hdfs" from the URI instead of hardcoding it.

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Removed the hardcoding for scheme "hdfs". Thanks!

yield Path(p)
jlowe marked this conversation as resolved.
Show resolved Hide resolved

def glob(path, pattern):
Expand Down
Loading