Merge pull request #309 from dclong/dev

Merge dev into main

Author: dclong

dsutil/__init__.py

@@ -3,4 +3,4 @@
 from . import git
 from . import poetry
 
-__version__ = "0.66.0"
+__version__ = "0.66.1"

dsutil/hadoop/log.py

@@ -289,6 +289,10 @@ def _error_priority(line: str) -> tuple[int, str, str]:
                 "(?i)NameError: name .* is not defined", 1,
                 "http://www.legendu.net/misc/blog/Spark-issue:-Pure-Python-code-errors"
             ),
+            (
+                r"(?i)TypeError: withReplacement", 1,
+                "http://www.legendu.net/misc/blog/spark-issue:-TypeError-withReplacement"
+            ),
             (
                 "(?i)table not found", 1,
                 "http://www.legendu.net/misc/blog/spark-issue-table-not-found/"

dsutil/hadoop/pyspark_submit.py

@@ -9,6 +9,7 @@
 import itertools as it
 from argparse import Namespace, ArgumentParser
 from pathlib import Path
+import tempfile
 import shutil
 import subprocess as sp
 import re
@@ -113,16 +114,14 @@ def _spark_log_filter(self, line: str) -> bool:
         return False
 
     @staticmethod
-    def _print_filter(line: str, log_filter: Union[Callable, None] = None) -> bool:
+    def _filter(line: str, time_begin, log_filter: Union[Callable, None] = None) -> str:
         if not line:
-            return False
-        if log_filter is None:
-            print(line)
-            return True
-        if log_filter(line):
-            print(line)
-            return True
-        return False
+            return ""
+        if log_filter is None or log_filter(line):
+            if "final status:" in line or " (state: " in line:
+                line = line + f" (Time elapsed: {datetime.datetime.now() - time_begin})"
+            return line
+        return ""
 
     def submit(self, cmd: str, attachments: Union[None, list[str]] = None) -> bool:
         """Submit a Spark job.
@@ -131,19 +130,25 @@ def submit(self, cmd: str, attachments: Union[None, list[str]] = None) -> bool:
         :param attachments: Attachments to send with the notification email.
         :return: True if the Spark application succeeds and False otherwise.
         """
-        logger.info("Submitting Spark job...\n{}", cmd)
+        time_begin = datetime.datetime.now()
+        logger.info("Submitting Spark job ...\n{}", cmd)
         stdout = []
         self._spark_submit_log.clear()
         process = sp.Popen(cmd, shell=True, stderr=sp.PIPE)
         while True:
             if process.poll() is None:
                 line = process.stderr.readline().decode().rstrip()  # pytype: disable=attribute-error
-                if self._print_filter(line, self._spark_log_filter):
+                line = self._filter(line, time_begin, self._spark_log_filter)
+                if line:
+                    print(line)
                     stdout.append(line)
             else:
                 for line in process.stderr.readlines():  # pytype: disable=attribute-error
-                    line = line.decode().rstrip()
-                    if self._print_filter(line, self._spark_log_filter):
+                    line = self._filter(
+                        line.decode().rstrip(), time_begin, self._spark_log_filter
+                    )
+                    if line:
+                        print(line)
                         stdout.append(line)
                 break
         # status
@@ -168,14 +173,21 @@ def submit(self, cmd: str, attachments: Union[None, list[str]] = None) -> bool:
                     attachments = [attachments]
                 if not isinstance(attachments, list):
                     attachments = list(attachments)
-                param["attachments"] = attachments
+                param["attachments"] = self._attach_txt(attachments)
             notifiers.get_notifier("email").notify(**param)
         if status == "FAILED":
             if self.email:
                 self._notify_log(app_id, "Re: " + subject)
             return False
         return True
 
+    @staticmethod
+    def _attach_txt(attachments: list[str]) -> list[str]:
+        dir_ = Path(tempfile.mkdtemp())
+        for attach in attachments:
+            shutil.copy2(attach, dir_)
+        return [str(path) + ".txt" for path in dir_.iterdir()]
+
     def _notify_log(self, app_id, subject):
         logger.info("Waiting for 300 seconds for the log to be available...")
         time.sleep(300)

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "dsutil"
-version = "0.66.0"
+version = "0.66.1"
 description = "A utils Python package for data scientists."
 authors = ["Benjamin Du <[email protected]>"]
 

readme.md

@@ -25,7 +25,7 @@ Currently, Python 3.7 and 3.8 are supported.
 
 You can download a copy of the latest release and install it using pip.
 ```bash
-pip3 install --user -U https://github.com/dclong/dsutil/releases/download/v0.66.0/dsutil-0.66.0-py3-none-any.whl
+pip3 install --user -U https://github.com/dclong/dsutil/releases/download/v0.66.1/dsutil-0.66.1-py3-none-any.whl
 ```
 Or you can use the following command to install the latest master branch
 if you have pip 20.0+.
@@ -35,7 +35,7 @@ pip3 install --user -U git+https://github.com/dclong/dsutil@main
 Use one of the following commands if you want to install all components of dsutil. 
 Available additional components are `cv`, `docker`, `pdf`, `jupyter`, `admin` and `all`.
 ```bash
-pip3 install "dsutil[cv] @ https://github.com/dclong/dsutil/releases/download/v0.66.0/dsutil-0.66.0-py3-none-any.whl"
+pip3 install "dsutil[cv] @ https://github.com/dclong/dsutil/releases/download/v0.66.1/dsutil-0.66.1-py3-none-any.whl"
 # or
 pip3 install --user -U "dsutil[all] @ git+https://github.com/dclong/dsutil@main"
 ```