Refactored FilesystemStore to use fsspec to support additional remote filesystems #2927

tgaddair · 2021-05-21T17:09:31Z

This adds support for various cloud storage systems like S3, GCS, ADLS, etc. in the Spark Estimators.

Fixes #2905.
Fixes #1732.

Signed-off-by: Travis Addair <tgaddair@gmail.com>

github-actions · 2021-05-21T18:55:07Z

Unit Test Results

    792 files ±0     792 suites ±0 6h 8m 13s ⏱️ ±0s
    600 tests ±0     564 ✔️ ±0     35 💤 ±0 0 ❌ ±0 1 🔥 ±0
16 473 runs ±0 12 439 ✔️ ±0 4 033 💤 ±0 0 ❌ ±0 1 🔥 ±0

For more details on these errors, see this check.

Results for commit 7a69711. ± Comparison against base commit 7a69711.

♻️ This comment has been updated with latest results.

umashankark · 2021-05-23T06:22:09Z

horovod/spark/common/store.py

+    @staticmethod
+    def _get_fs_and_protocol(url):
+        protocol, path = split_protocol(url)
+        fs = fsspec.filesystem(protocol)


Can storage_options be provided as second argument [fsspec.filesystem(protocol, **storage_options)] ? This will help provide authentication options to fsspec.

umashankark · 2021-05-23T06:40:35Z

horovod/spark/common/store.py


    def __init__(self, prefix_path, *args, **kwargs):
-        self._fs = pa.LocalFileSystem()
-        super(LocalStore, self).__init__(prefix_path, *args, **kwargs)
+        self._fs, self.protocol = FilesystemStore._get_fs_and_protocol(prefix_path)


Can this be added ?:
self.storage_options = kwargs['storage_options'] if 'storage_options' in kwargs else dict()
self._fs, self.protocol = FilesystemStore._get_fs_and_protocol(prefix_path, self.storage_options)

umashankark · 2021-05-23T15:24:56Z

Also, adding storage_options dict param to HorovodEstimator (EstimatorParams ?), should help RemoteTrainer()s to pass them to make_reader & make_batch_reader Petastorm calls.

umashankark · 2021-05-27T15:50:11Z

I am testing this change with ADLS; I will update with results.

tgaddair · 2021-05-28T17:56:42Z

Thanks @umashankark, I hope to be able to follow-up on your suggestions soon (maybe this weekend). Just got a few other things I need to take care of first. Thanks for your patience!

umashankark · 2021-05-29T10:34:19Z

I generated xor dataset using spark_common.create_xor_data() and copied to ADLS.
Took "xor" model estimator (from testcases).
Used FsspecStore with minor changes (added_storage_options, and used same storage_options for make_reader,make_batch_reader access in remote.py) .
Code:
store = Store.create('abfs://container/', storage_options=storage_options,
train_path='abfs://container/xor-data', save_runs=True)
torch_estimator = hvd_spark.TorchEstimator(
num_proc=1,
store=store,
model=model,
optimizer=optimizer,
loss=loss,
input_shapes=[[1],[1]],
feature_cols=['x1','x2'],
label_cols=['y'],
batch_size=1,
epochs=10,
verbose=1,)
torch_model = torch_estimator.fit_on_parquet()

Getting this error only when Estimator(save_runs=True):
[0]: File "/home/lll/tests/final-hvd/venv/lib/python3.8/site-packages/horovod-0.22.0-py3.8-linux-x86_64.egg/horovod/spark/torch/remote.py", line 401, in train
[0]: remote_store.sync(run_output_dir)
[0]: File "/home/lll/tests/final-hvd/venv/lib/python3.8/site-packages/horovod-0.22.0-py3.8-linux-x86_64.egg/horovod/spark/common/store.py", line 307, in fn
[0]: self.fs.put(local_run_path, run_path, recursive=True)
[0]: File "/home/lll/tests/final-hvd/venv/lib/python3.8/site-packages/fsspec/asyn.py", line 72, in wrapper
[0]: return sync(self.loop, func, *args, **kwargs)
[0]: File "/home/lll/tests/final-hvd/venv/lib/python3.8/site-packages/fsspec/asyn.py", line 53, in sync
[0]: raise result[0]
[0]: File "/home/lll/tests/final-hvd/venv/lib/python3.8/site-packages/fsspec/asyn.py", line 20, in _runner
[0]: result[0] = await coro
[0]: File "/home/lll/tests/final-hvd/venv/lib/python3.8/site-packages/fsspec/asyn.py", line 283, in _put
[0]: return await _throttled_gather(
[0]: File "/home/lll/tests/final-hvd/venv/lib/python3.8/site-packages/fsspec/asyn.py", line 144, in _throttled_gather
[0]: results.extend(await asyncio.gather(*chunk, **gather_kwargs))
[0]: File "/home/lll/tests/final-hvd/venv/lib/python3.8/site-packages/adlfs/spec.py", line 1432, in _put_file
[0]: raise FileExistsError("File already exists!")
[0]:FileExistsError: File already exists!

This happens in second epoch. Looks like Store.sync_fn:: fs.put() couldn't overwrite the folder content. Checking whether fs.put() expected to overwrite.

umashankark · 2021-05-29T11:10:32Z

FilesystemStore:

Store.sync_fn():
     """ snip
        self.fs.put(local_run_path, run_path, recursive=True, overwrite=True)

overwrite=True takes care of the error. Will share the minor changes done.

umashankark · 2021-05-29T17:12:07Z

Tested and pushed changes to #2947. This is PR against fsspec branch. @tgaddair

Signed-off-by: Travis Addair <tgaddair@gmail.com>

tgaddair · 2021-06-02T21:59:44Z

@chongxiaoc @irasit please take a look. Should hopefully not break anything on the Uber side.

chongxiaoc

LGTM.

…l remote filesystems (horovod#2927)" This reverts commit 7a69711.

…l remote filesystems (horovod#2927)" This reverts commit 7a69711. Signed-off-by: Chongxiao Cao <chongxiaoc@uber.com>

tgaddair mentioned this pull request May 21, 2021

Supporting ADLS storage #2905

Closed

tgaddair added 2 commits May 21, 2021 10:31

WIP

c38e32e

Refactored LocalStore into FilesystemStore

bdaa5f2

Signed-off-by: Travis Addair <tgaddair@gmail.com>

tgaddair force-pushed the fsspec branch from 95fa3ca to bdaa5f2 Compare May 21, 2021 17:31

umashankark reviewed May 23, 2021

View reviewed changes

umashankark and others added 3 commits June 1, 2021 17:20

added storage_options to FilesystemStore (#2947)

9a00a70

Merged master

8ab2751

Fixed docs

7f63261

Signed-off-by: Travis Addair <tgaddair@gmail.com>

tgaddair requested review from irasit and chongxiaoc June 2, 2021 21:59

chongxiaoc approved these changes Jun 2, 2021

View reviewed changes

tgaddair merged commit 7a69711 into master Jun 2, 2021

tgaddair deleted the fsspec branch June 2, 2021 22:22

chongxiaoc added a commit to chongxiaoc/horovod that referenced this pull request Jun 17, 2021

Revert "Refactored FilesystemStore to use fsspec to support additiona…

4d74e7b

…l remote filesystems (horovod#2927)" This reverts commit 7a69711.

tgaddair mentioned this pull request Jun 27, 2021

Removed store from closure variables sent to remote workers #3004

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Refactored FilesystemStore to use fsspec to support additional remote filesystems #2927

Refactored FilesystemStore to use fsspec to support additional remote filesystems #2927

tgaddair commented May 21, 2021 •

edited

github-actions bot commented May 21, 2021 •

edited

umashankark May 23, 2021

umashankark May 23, 2021

umashankark commented May 23, 2021

umashankark commented May 27, 2021

tgaddair commented May 28, 2021

umashankark commented May 29, 2021 •

edited

umashankark commented May 29, 2021 •

edited

umashankark commented May 29, 2021

tgaddair commented Jun 2, 2021

chongxiaoc left a comment

Refactored FilesystemStore to use fsspec to support additional remote filesystems #2927

Refactored FilesystemStore to use fsspec to support additional remote filesystems #2927

Conversation

tgaddair commented May 21, 2021 • edited

github-actions bot commented May 21, 2021 • edited

Unit Test Results

umashankark May 23, 2021

Choose a reason for hiding this comment

umashankark May 23, 2021

Choose a reason for hiding this comment

umashankark commented May 23, 2021

umashankark commented May 27, 2021

tgaddair commented May 28, 2021

umashankark commented May 29, 2021 • edited

umashankark commented May 29, 2021 • edited

umashankark commented May 29, 2021

tgaddair commented Jun 2, 2021

chongxiaoc left a comment

Choose a reason for hiding this comment

tgaddair commented May 21, 2021 •

edited

github-actions bot commented May 21, 2021 •

edited

umashankark commented May 29, 2021 •

edited

umashankark commented May 29, 2021 •

edited