Repository: horovod/horovod
Pull request: Fix Horovod pyarrow IndexError: list index out of range
## Checklist before submitting

- [x] Did you read the [contributor guide](https://github.com/horovod/horovod/blob/master/CONTRIBUTING.md)?
- [ ] Did you update the docs?
- [ ] Did you write any tests to validate this change?  
- [ ] Did you update the [CHANGELOG](https://github.com/horovod/horovod/blob/master/CHANGELOG.md), if this change affects users?

## Description

Fixes #2193 .
Fix Horovod pyarrow IndexError: list index out of range

## Review process to land 

1. All tests and other checks must succeed.
2. At least one member of the [technical steering committee](https://github.com/horovod/horovod/blob/master/CONTRIBUTING.md) must review and approve.
3. If any member of the technical steering committee requests changes, they must be addressed.


Problem statement:
Horovod pyarrow IndexError: list index out of range
**Environment:**
1. Framework: TensorFlow
2. Framework version: 1.14
3. Horovod version: 0.19.5
4. MPI version: 4.0.4
5. CUDA version: NA
6. NCCL version: NA
7. Python version: 3.6
8. Spark / PySpark version: 2.4.4
9. OS and version: Amazon Linux
10. GCC version: 7.2.1
11. pyarrow version: 0.15.1


The nmist example. error is on the model fitting:
```
>>> #store = Store.create("hdfs:///user/hvd")
... # Train a Horovod Spark Estimator on the DataFrame
... keras_estimator = hvd.KerasEstimator(num_proc=1,
...                                      store=store,
...                                      model=model,
...                                      optimizer=optimizer,
...                                      loss=loss,
...                                      metrics=['accuracy'],
...                                      feature_cols=['features'],
...                                      label_cols=['label_vec'],
...                                      batch_size=256,
...                                      epochs=10,
...                                      verbose=3)
>>>
>>> keras_model = keras_estimator.fit(train_df).setOutputCols(['label_prob'])
num_partitions=10
writing dataframes
train_data_path=file:///var/lib/hvd/intermediate_train_data.0
val_data_path=file:///var/lib/hvd/intermediate_val_data.0
train_partitions=10
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/home/hadoop/conda/envs/py36/lib/python3.6/site-packages/horovod/spark/common/estimator.py", line 37, in fit
    return super(HorovodEstimator, self).fit(df, params)
  File "/usr/lib/spark/python/pyspark/ml/base.py", line 132, in fit
    return self._fit(dataset)
  File "/home/hadoop/conda/envs/py36/lib/python3.6/site-packages/horovod/spark/common/estimator.py", line 78, in _fit
    verbose=self.getVerbose()) as dataset_idx:
  File "/home/hadoop/conda/envs/py36/lib/python3.6/contextlib.py", line 81, in __enter__
    return next(self.gen)
  File "/home/hadoop/conda/envs/py36/lib/python3.6/site-packages/horovod/spark/common/util.py", line 637, in prepare_data
    num_partitions, num_processes, verbose)
  File "/home/hadoop/conda/envs/py36/lib/python3.6/site-packages/horovod/spark/common/util.py", line 573, in _get_or_create_dataset
    store, label_columns, feature_columns, sample_weight_col, dataset_idx)
  File "/home/hadoop/conda/envs/py36/lib/python3.6/site-packages/horovod/spark/common/util.py", line 457, in get_simple_meta_from_parquet
    train_data = store.get_parquet_dataset(train_data_path)
  File "/home/hadoop/conda/envs/py36/lib/python3.6/site-packages/horovod/spark/common/store.py", line 176, in get_parquet_dataset
    return pq.ParquetDataset(self.get_localized_path(path), filesystem=self.get_filesystem())
  File "/home/hadoop/conda/envs/py36/lib/python3.6/site-packages/pyarrow/parquet.py", line 1060, in __init__
    self.validate_schemas()
  File "/home/hadoop/conda/envs/py36/lib/python3.6/site-packages/pyarrow/parquet.py", line 1092, in validate_schemas
    self.schema = self.pieces[0].get_metadata().schema
IndexError: list index out of range
```

I have 1 master node and 1 worker node and the Spark is running in the client mode. 
When I checked the `intermediate_train_data.0` folder. In the master node, it has only a `_SUCCESS` file; In the worker node, it has `_temporary` file.  It seems the writing process hasn't been completed yet in this worker node before the file being read. 
There are no `intermediate_val_data` in both nodes.

Hints from issue discussion:
Hey @ZhenyiLin, I believe Databricks ran into this issue when implementing the Spark Dataset Converter for Petastorm (https://github.com/uber/petastorm/pull/496).  

We have a PR in progress to migrate to the Spark Dataset Converter API, which should hopefully resolve this issue: #2091.

@WeichenXu123, can you confirm that the Spark Dataset Converter should  avoid this race condition?
@tgaddair  @WeichenXu123  
Thanks for the technical support. We are configing and testing Horovod in AWS EMR.

Some more information to share:
1) The error makes the Horovod run unstable. It is Not always failed. Sometimes it works fine but sometimes not
2) a) By repartitioning the Spark Dataframe into smaller partitions; b) changing Spark resources may help work round, but the error seems to appear randomly and I have to manually try it.

  


Hi, @tgaddair @WeichenXu123 I have exactly the same issue in running keras_spark_mnist.py on a spark cluster. Following is what I see.
"I want to inform you that in my situation '_SUCCESS' file is created in all workers directories "


>`writing dataframes train_data_path=file:///home/spark/sample_code/horovod_keras_spark_mnist/intermediate_train_data.0 val_data_path=file:///home/spark/sample_code/horovod_keras_spark_mnist/intermediate_val_data.0 train_partitions=80 Traceback (most recent call last): File "keras_spark_mnist.py", line 125, in <module> keras_model = keras_estimator.fit(train_df).setOutputCols(['label_prob']) File "/home/spark/anaconda3/lib/python3.8/site-packages/horovod/spark/common/estimator.py", line 35, in fit return super(HorovodEstimator, self).fit(df, params) File "/home/spark/anaconda3/lib/python3.8/site-packages/pyspark/ml/base.py", line 161, in fit return self._fit(dataset) File "/home/spark/anaconda3/lib/python3.8/site-packages/horovod/spark/common/estimator.py", line 68, in _fit with util.prepare_data(backend.num_processes(), File "/home/spark/anaconda3/lib/python3.8/contextlib.py", line 113, in __enter__ return next(self.gen) File "/home/spark/anaconda3/lib/python3.8/site-packages/horovod/spark/common/util.py", line 668, in prepare_data dataset_idx = _get_or_create_dataset(key, store, df, feature_columns, label_columns, File "/home/spark/anaconda3/lib/python3.8/site-packages/horovod/spark/common/util.py", line 605, in _get_or_create_dataset train_rows, val_rows, pq_metadata, avg_row_size = get_simple_meta_from_parquet( File "/home/spark/anaconda3/lib/python3.8/site-packages/horovod/spark/common/util.py", line 490, in get_simple_meta_from_parquet train_data = store.get_parquet_dataset(train_data_path) File "/home/spark/anaconda3/lib/python3.8/site-packages/horovod/spark/common/store.py", line 213, in get_parquet_dataset return pq.ParquetDataset(self.get_localized_path(path), filesystem=self.fs) File "/home/spark/anaconda3/lib/python3.8/site-packages/pyarrow/parquet.py", line 1364, in __init__ self.validate_schemas() File "/home/spark/anaconda3/lib/python3.8/site-packages/pyarrow/parquet.py", line 1395, in validate_schemas self.schema = self._pieces[0].get_metadata().schema IndexError: list index out of range'

Commit message: update

Signed-off-by: Weichen Xu <weichen.xu@databricks.com>

Patch to review:
diff --git a/horovod/spark/common/util.py b/horovod/spark/common/util.py
index ae2a41a261..1a1dc604c0 100644
--- a/horovod/spark/common/util.py
+++ b/horovod/spark/common/util.py
@@ -17,7 +17,9 @@
 
 import contextlib
 import os
+import time
 
+from multiprocessing.pool import ThreadPool
 import pyarrow as pa
 import numpy as np
 import pyspark.sql.functions as f
@@ -30,6 +32,8 @@
 except ImportError:
     from pyspark.sql.types import from_arrow_type
 
+from pyspark.sql import SparkSession
+
 from horovod.runner.common.util import codec, host_hash as hh
 from horovod.spark.common import cache, constants
 
@@ -539,6 +543,46 @@ def _train_val_split(df, validation):
     return train_df, val_df, validation_ratio
 
 
+_FILE_AVAILABILITY_WAIT_TIMEOUT_SECS = \
+    int(os.environ.get('FILE_AVAILABILITY_WAIT_TIMEOUT_SECS', '30'))
+
+
+def _wait_file_available(store, url_list):
+    """Waiting about _FILE_AVAILABILITY_WAIT_TIMEOUT_SECS seconds (default 30 seconds) to make sure
+    all files are available for reading. This is useful in some filesystems, such as S3 which only
+    providing eventually consistency.
+    """
+    # Import LocalStore here to avoid circular import
+    from horovod.spark.common.store import LocalStore
+    if isinstance(store, LocalStore):
+        return
+
+    def wait_for_file(path):
+        end_time = time.time() + _FILE_AVAILABILITY_WAIT_TIMEOUT_SECS
+        while time.time() < end_time:
+            if store.exists(path):
+                return True
+            time.sleep(0.1)
+        return False
+
+    pool = ThreadPool(min(len(url_list), 64))
+    try:
+        results = pool.map(wait_for_file, url_list)
+        failed_list = [url for url, result in zip(url_list, results) if not result]
+        if failed_list:
+            raise RuntimeError('Timeout while waiting for all parquet-store files to appear at urls {failed_list},'
+                               'Please check whether these files were saved successfully when materializing dataframe.'
+                               .format(failed_list=','.join(failed_list)))
+    finally:
+        pool.close()
+        pool.join()
+
+
+def get_spark_df_saved_file_list(saved_path):
+    spark_session = SparkSession.builder.getOrCreate()
+    return list(spark_session.read.parquet(saved_path)._jdf.inputFiles())
+
+
 def _get_or_create_dataset(key, store, df, feature_columns, label_columns,
                            validation, sample_weight_col, compress_sparse,
                            num_partitions, num_processes, verbose):
@@ -590,6 +634,8 @@ def _get_or_create_dataset(key, store, df, feature_columns, label_columns,
                 .mode('overwrite') \
                 .parquet(train_data_path)
 
+            saved_file_list = get_spark_df_saved_file_list(train_data_path)
+
             if val_df:
                 val_partitions = max(int(num_partitions * validation_ratio),
                                      num_processes)
@@ -602,6 +648,10 @@ def _get_or_create_dataset(key, store, df, feature_columns, label_columns,
                     .mode('overwrite') \
                     .parquet(val_data_path)
 
+                saved_file_list += get_spark_df_saved_file_list(val_data_path)
+
+            _wait_file_available(store, saved_file_list)
+
             train_rows, val_rows, pq_metadata, avg_row_size = get_simple_meta_from_parquet(
                 store, label_columns, feature_columns, sample_weight_col, dataset_idx)
 
diff --git a/test/utils/spark_common.py b/test/utils/spark_common.py
index feb23c8023..e0a4ebaba0 100644
--- a/test/utils/spark_common.py
+++ b/test/utils/spark_common.py
@@ -16,8 +16,11 @@
 import contextlib
 import os
 import platform
+import pytest
 import stat
 import sys
+import threading
+import time
 
 from tempfile import TemporaryDirectory
 
@@ -30,6 +33,7 @@
 
 from horovod.runner.common.util import secret
 from horovod.spark.common.store import LocalStore
+from horovod.spark.common.util import _wait_file_available
 from horovod.spark.driver.driver_service import SparkDriverService, SparkDriverClient
 from horovod.spark.task.task_service import SparkTaskService, SparkTaskClient
 
@@ -232,3 +236,41 @@ def create_mnist_data(spark):
 
 def create_test_data_from_schema(spark, data, schema):
     return spark.createDataFrame(data, schema=schema)
+
+
+def test_wait_file_available():
+    with tempdir() as d:
+        pq_dir = os.path.join(d, 'test_ev')
+        os.makedirs(pq_dir)
+        file1_path = os.path.join(pq_dir, 'file1')
+        file2_path = os.path.join(pq_dir, 'file2')
+        url1 = 'file://' + file1_path.replace(os.sep, '/')
+        url2 = 'file://' + file2_path.replace(os.sep, '/')
+
+        url_list = [url1, url2]
+
+        def create_file(p):
+            with open(p, 'w'):
+                pass
+
+        # 1. test all files exists.
+        create_file(file1_path)
+        create_file(file2_path)
+        _wait_file_available(url_list)
+
+        # 2. test one file does not exists. Raise error.
+        os.remove(file2_path)
+        with pytest.raises(
+                RuntimeError,
+                match='Timeout while waiting for all parquet-store files to appear at urls'
+        ):
+            _wait_file_available(url_list)
+
+        # 3. test one file accessible after 1 second.
+        def delay_create_file2():
+            time.sleep(1)
+            create_file(file2_path)
+
+        threading.Thread(target=delay_create_file2()).start()
+
+        _wait_file_available(url_list)


Code review comments:
diff_hunk: @@ -539,6 +543,46 @@ def _train_val_split(df, validation):
     return train_df, val_df, validation_ratio
 
 
+_FILE_AVAILABILITY_WAIT_TIMEOUT_SECS = \
+    int(os.environ.get('FILE_AVAILABILITY_WAIT_TIMEOUT_SECS', '30'))
+
+
+def _wait_file_available(store, url_list):
+    """Waiting about _FILE_AVAILABILITY_WAIT_TIMEOUT_SECS seconds (default 30 seconds) to make sure
+    all files are available for reading. This is useful in some filesystems, such as S3 which only
+    providing eventually consistency.
+    """
+    # Import LocalStore here to avoid circular import
+    from horovod.spark.common.store import LocalStore
+    if isinstance(store, LocalStore):
+        return
+
+    def wait_for_file(path):
+        end_time = time.time() + _FILE_AVAILABILITY_WAIT_TIMEOUT_SECS
+        while time.time() < end_time:
+            if store.exists(path):
+                return True
+            time.sleep(0.1)
+        return False
+
+    pool = ThreadPool(min(len(url_list), 64))
+    try:
+        results = pool.map(wait_for_file, url_list)
+        failed_list = [url for url, result in zip(url_list, results) if not result]
+        if failed_list:
+            raise RuntimeError('Timeout while waiting for all parquet-store files to appear at urls {failed_list},'
+                               'Please check whether these files were saved successfully when materializing dataframe.'
+                               .format(failed_list=','.join(failed_list)))
+    finally:
+        pool.close()
+        pool.join()
+
+
+def get_spark_df_saved_file_list(saved_path):
path: horovod/spark/common/util.py
text: @user1:
can we also have a test for this method?

@user1:
I think this method should be private, so `_get_spark_df_saved_file_list`

@author:
updated.