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.

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

[fb] Improve WebHDFS read performance by always making request for 1MB during read #588

Closed
wants to merge 1 commit into from
Closed

[fb] Improve WebHDFS read performance by always making request for 1MB during read #588

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
20 changes: 18 additions & 2 deletions desktop/libs/hadoop/src/hadoop/fs/webhdfs.py
View file
Open in desktop
Expand Up @@ -835,6 +835,7 @@ def __init__(self, fs, path, mode='r'):
self._path = normpath(path)
self._pos = 0
self._mode = mode
self._data = ""

try:
self._stat = fs.stats(path)
Expand All @@ -859,6 +860,9 @@ def seek(self, offset, whence=0):
else:
raise IOError(errno.EINVAL, _("Invalid argument to seek for whence"))

# Reset cached data on seek
self._data = ""

def stat(self):
self._stat = self._fs.stats(self._path)
return self._stat
Expand All @@ -867,8 +871,20 @@ def tell(self):
return self._pos

def read(self, length=DEFAULT_READ_SIZE):
data = self._fs.read(self._path, self._pos, length)
self._pos += len(data)
# Make sure to grab more data if there is none available
if length > len(self._data):
# Grab at minimum DEFAULT_READ_SIZE
self._data += self._fs.read(self._path, self._pos, max(length, DEFAULT_READ_SIZE))

# Make sure to increase the _pos by length of data or what length of requested
real_length = min(len(self._data), length)
self._pos += real_length

# Strip off real length from data set
data = self._data[:real_length]
self._data = self._data[real_length:]

# Return the requested length of data
return data

def write(self, data):
Expand Down