Update _localspark_submission.py (#973)

feathr_project/feathr/spark_provider/_localspark_submission.py

@@ -178,14 +178,20 @@ def wait_for_completion(self, timeout_seconds: Optional[float] = 500) -> bool:
         if proc.returncode == None:
             logger.warning(
                 f"Spark job with pid {self.latest_spark_proc.pid} not completed after {timeout_seconds} sec \
-                    time out setting. Please check."
+                    time out setting. Spark Logs:"
             )
+            with open(log_read.name) as f:
+                contents = f.read()
+                logger.error(contents)
             if self.clean_up:
                 self._clean_up()
                 proc.wait()
                 return True
         elif proc.returncode == 1:
-            logger.warning(f"Spark job with pid {self.latest_spark_proc.pid} is not successful. Please check.")
+            logger.warning(f"Spark job with pid {self.latest_spark_proc.pid} is not successful. Spark Logs:")
+            with open(log_read.name) as f:
+                contents = f.read()
+                logger.error(contents)
             return False
         else:
             logger.info(
@@ -247,7 +253,7 @@ def _get_debug_file_name(self, debug_folder: str = "debug", prefix: str = None):
         prefix += datetime.now().strftime("%Y%m%d%H%M%S")
         debug_path = os.path.join(debug_folder, prefix)
 
-        print(debug_path)
+        logger.info(f"Spark log path is {debug_path}")
         if not os.path.exists(debug_path):
             os.makedirs(debug_path)