Open tadas-subonis opened 3 years ago
selitvin
commented
3 years ago The error indicates that we were trying to reset() a reader before the number of epochs specified during its creation was consumed. It's hard to say what went wrong without seeing how do you create the reader/dataloader. Can you please post that code?
Also, please post the version of pytorch_lightning that you use.
tadas-subonis
commented
3 years ago petastorm[torch] (0.9.5)
pytorch-lightning==0.9.0Creation code (a bit simplified)
indexed_dataset = some_data \
.zipWithUniqueId()
def is_validation(x):
return (x[1] % 100) >= 80
def not_(f):
def step(x):
return not f(x)
return step
validation_dataset = indexed_dataset \
.filter(is_validation) \
.map(lambda x: x[0])
training_dataset = indexed_dataset \
.filter(not_(is_validation)) \
.map(lambda x: x[0])
def rdd_to_df(rdd):
rdd = rdd.map(lambda x: Row(**(x)))
return spark.createDataFrame(rdd)
training_dataset_rdd = training_dataset.map(vectorize).map(create_pytorch_dict_with_numpy_bytes)
validation_dataset_rdd = validation_dataset.map(vectorize).map(create_pytorch_dict_with_numpy_bytes)
converter_train = make_spark_converter(rdd_to_df(training_dataset_rdd))
converter_test = make_spark_converter(rdd_to_df(validation_dataset_rdd))
For now, I've loaded the validation dataset into memory and used the regular DataLoader and it seems to work fine.
But I've also noticed a strange behavior just recently - if I start training from "scratch", the process will train as it is supposed to - it will go from epoch 1 to 50. However, if I reuse the same session run and run the pipeline without modifications, it will run just the last epoch and terminate.
@selitvin do you have any ideas why that could happen? :)
Thanks
selitvin
commented
3 years ago Hmm, I am not sure. I am not familiar with pytorch-lighting and how they interact with DataLoader.
LeonardoEmili
commented
3 years ago I am facing the same issue using a custom petastorm.Dataloader (actually I only provide a custom collate function, not changing the Dataloader at all) and feeding it to pytorch-lightning Trainer. @tadas-subonis were you able to fix it or we shall rely on the default Dataloader if using pytorch-lightning? By chance, did you experiment with PyTorch to see if in that case, it works as expected?
darwinharianto
commented
3 years ago Is there any progress on this?
LeonardoEmili
commented
3 years ago Is there any progress on this?
I had to switch to plain PyTorch, was not able to do it using Pytorch Lightning,
maxisawesome
commented
2 years ago Also getting this issue. At first I was using "limit_train_batches" in my Pytorch Lightning trainer, but I removed it and still ended up with this error. PTL runs two validation steps at the beginning of the model, then returns to the validation dataloader when it finishes a training epoch. I assume, upon beginning the full validation run, it resets the val dataloader. If this is what's happening, it sounds like Pytorch Lightning is not supported. If so, that should be made clear somewhere.
Unfortunately, I cannot copy my stack trace due to some VPN things, but the error occurs at line 116 in petastorm/pytorch.py and then line 490 at petastorm/reader.py
maxisawesome
commented
2 years ago I solved this w/ num_sanity_val_steps=0 in my PytorchLightning Trainer! Seems like my intuition above was correct, and after turning the sanity check off, the val set doesnt need be reset, and no errors occur.
Nice!
wbeardall
commented
1 year ago As an alternative fix to this, there's a relatively simple workaround that involves subclassing the Petastorm DataLoader, such that instead of being initialized with a Reader object, it is initialized with a zero-argument function which returns a Reader, so that the loader can re-create its reader if needs be. A possible implementation of this is given below:
import logging
from petastorm.pytorch import _PARALLEL_ITER_ERROR, DataLoader, decimal_friendly_collate
logger = logging.getLogger(__name__)
class SafeResetLoader(DataLoader):
"""
A data loader adaptor for ``torch.utils.data.DataLoader``.
This class iterates and returns items from the Reader in batches.
Behaves the same as the Petastorm dataloader, except it takes a function which generates a reader,
so that it can reset itself at any time.
This loader can be used as an iterator and will terminate when the reader used in the construction of the class
runs out of samples.
"""
def __init__(self, reader_fn, batch_size=1, collate_fn=decimal_friendly_collate,
shuffling_queue_capacity=0):
"""
Initializes a data loader object, with a default collate.
Number of epochs is defined by the configuration of the reader argument.
An optional shuffling queue is created if shuffling_queue_capacity is greater than 0. No samples will be
returned to a user by the ``DataLoader`` until the queue is full. After that, batches of `batch_size`
will be created by uniformly sampling the shuffling queue. Once no more samples are available from the data
reader, the shuffling queue is allowed to be consumed till no further samples are available.
Note that the last returned batch could have less then ``batch_size`` samples.
NOTE: ``make_batch_reader`` has it's own ``shuffle_row_groups`` argument. It randomizes order in
which parquet row-groups are loaded and has no effect on the order of rows within each row-group. To achieve
row-level shuffling you should set shuffling_queue_capacity to a non zero value.
:param reader: petastorm Reader instance
:param batch_size: the number of items to return per batch; factored into the len() of this reader
:param collate_fn: an optional callable to merge a list of samples to form a mini-batch.
:param shuffling_queue_capacity: Queue capacity is passed to the underlying :class:`tf.RandomShuffleQueue`
instance. If set to 0, no shuffling will be done.
"""
super().__init__(reader_fn(),batch_size=batch_size,collate_fn=collate_fn,
shuffling_queue_capacity=shuffling_queue_capacity)
self._reader_fn = reader_fn
def __iter__(self):
if self._error is not None:
raise RuntimeError('Cannot start a new iteration because last time iteration failed with error {err}.'
.format(err=repr(self._error)))
if self._in_iter is not None and self._in_iter == True: # noqa: E712
raise RuntimeError(_PARALLEL_ITER_ERROR)
if self._in_iter is not None:
try:
self.reader.reset()
except NotImplementedError:
# Called reset in the middle of an iteration.
self.reader.stop()
self.reader.join()
self.reader = self._reader_fn()
logger.warning('Start a new pass of Petastorm DataLoader, reset underlying Petastorm reader to position 0.')
self._in_iter = True
try:
for batch in self._iter_impl():
yield batch
except Exception as e:
self._error = e
logger.error('Iteration on Petastorm DataLoader raise error: %s', repr(e))
raise
finally:
self._in_iter = False
def terminate(self):
self.reader.stop()
self.reader.join()
This loader can then be called in the following manner:
import functools
make_reader_fn = functools.partial(partial(make_reader, ...))
loader = SafeResetLoader(make_reader_fn, collate_fn=..., batch_size=...)
elbamos
commented
1 month ago Is there any progress on this issue? Its a pretty big gap if petastorm isn't compatible with pytorch lightning.
selitvin
commented
1 month ago @elbamos : I am not actively maintaining petastorm any longer. I am not aware of any active contributors at this stage.
I have this piece of code:
when it gets to the validation bit, it throws the previously mentioned
I am using PyTorch Lightning
The stack looks like this:
Lightning trainer config:
Is there something that I am doing wrong?