lightning: turn off shuffling for validation dataset

Signed-off-by: Chongxiao Cao <chongxiaoc@uber.com>

horovod/spark/lightning/remote.py

@@ -172,11 +172,11 @@ def train(serialized_model):
             with set_data_loader(model, remote_store.train_data_path, 'train_dataloader',
                                  train_reader_worker_count, reader_pool_type,
                                  name="train_dataloader",
-                                 limit_step_per_epoch=_train_steps_per_epoch), \
+                                 limit_step_per_epoch=_train_steps_per_epoch, is_val=False), \
                     set_data_loader(model, remote_store.val_data_path, 'val_dataloader',
                                     val_reader_worker_count, reader_pool_type,
                                     should_validate, name="val_dataloader",
-                                    limit_step_per_epoch=_val_steps_per_epoch):
+                                    limit_step_per_epoch=_val_steps_per_epoch, is_val=True):
 
                 trainer.fit(model)
 
@@ -222,7 +222,8 @@ def on_sanity_check_end(self, trainer, model):
 def _set_data_loader_fn(transformation, schema_fields, batch_size, calculate_shuffle_buffer_size, data_loader_cls, num_epochs, store, verbose=False):
 
     @contextlib.contextmanager
-    def set_data_loader(model, data_path, dataloader_attr, reader_worker_count, reader_pool_type, should_read=True, name="", limit_step_per_epoch=-1):
+    def set_data_loader(model, data_path, dataloader_attr, reader_worker_count, reader_pool_type,
+                        should_read=True, name="", limit_step_per_epoch=-1, is_val=False):
         from petastorm import TransformSpec, make_reader, make_batch_reader
         import horovod.torch as hvd
 
@@ -268,7 +269,7 @@ def set_data_loader(model, data_path, dataloader_attr, reader_worker_count, read
                             **reader_factory_kwargs) as reader:
             def dataloader_fn():
                 return data_loader_cls(reader=reader, batch_size=batch_size,
-                                       shuffling_queue_capacity=calculate_shuffle_buffer_size(),
+                                       shuffling_queue_capacity=0 if is_val else calculate_shuffle_buffer_size(),
                                        name=name,
                                        limit_step_per_epoch=limit_step_per_epoch,
                                        verbose=verbose)