Wait for schema in distribution

astrolabsoftware · Dec 13, 2024 · 935cb91 · 935cb91
1 parent 72f1353
commit 935cb91
Showing 1 changed file with 11 additions and 1 deletion.
diff --git a/fink_broker/spark_utils.py b/fink_broker/spark_utils.py
@@ -336,7 +336,17 @@ def connect_to_raw_database(basepath: str, path: str, latestfirst: bool) -> Data
         wait_sec = increase_wait_time(wait_sec)
 
     # Create a DF from the database
-    userschema = spark.read.parquet(basepath).schema
+    # We need to wait for the schema to be available
+    while True:
+        try:
+            userschema = spark.read.parquet(basepath).schema
+        except Exception as e:
+            _LOG.error("Error while reading %s, %s", basepath, e)
+            time.sleep(wait_sec)
+            wait_sec = increase_wait_time(wait_sec)
+            continue
+        else:
+            break
 
     df = (
         spark.readStream.format("parquet")