Merge remote-tracking branch 'origin/develop'

Author: awdeorio

README_Hadoop_Streaming.md

@@ -289,7 +289,14 @@ if __name__ == "__main__":
 ## Tips and tricks
 These are some pro-tips for working with MapReduce programs written in Python for the Hadoop Streaming interface.
 
-### Debugging
+### Print debug messages to `stderr`
+To avoid interfering with pipeline output in `stdout`, direct debugging print messages to `stderr`:
+
+```python
+print("DEBUG finding bugs... ~(^._.)", file=sys.stderr)
+```
+
+### Debugging with PDB
 We encounter a problem if we add a `breakpoint()` in `map.py`.
 ```python
 for line in sys.stdin:
@@ -299,13 +306,13 @@ for line in sys.stdin:
         print(f"{word}\t1")
 ```
 
-PDB/PDB++ confuses the stdin being piped in from the input file for user input, so we get these errors:
+PDB confuses the stdin being piped in from the input file for user input, so we get these errors:
 ```console
 $ cat input/input* | ./map.py
 ...
-(Pdb++) *** SyntaxError: invalid syntax
-(Pdb++) *** SyntaxError: invalid syntax
-(Pdb++) *** SyntaxError: invalid syntax
+(Pdb) *** SyntaxError: invalid syntax
+(Pdb) *** SyntaxError: invalid syntax
+(Pdb) *** SyntaxError: invalid syntax
 ...
 ```
 
@@ -325,7 +332,7 @@ Now our debugger works correctly.
 ```console
 $ cat input/input* | ./map.py
 ...
-(Pdb++)
+(Pdb)
 ```
 
 Don't forget to remove your temporary changes!

madoop/mapreduce.py

@@ -43,6 +43,8 @@ def mapreduce(
     # Executable scripts must have valid shebangs
     is_executable(map_exe)
     is_executable(reduce_exe)
+    if partitioner:
+        is_executable(partitioner)
 
     # Create a tmp directory which will be automatically cleaned up
     with tempfile.TemporaryDirectory(prefix="madoop-") as tmpdir:
@@ -173,7 +175,13 @@ def is_executable(exe):
             stderr=subprocess.PIPE,
             check=True,
         )
-    except (subprocess.CalledProcessError, OSError) as err:
+    except subprocess.CalledProcessError as err:
+        raise MadoopError(
+            f"Failed executable test: {err}"
+            f"\n{err.stdout.decode()}" if err.stdout else ""
+            f"{err.stderr.decode()}" if err.stderr else ""
+        ) from err
+    except OSError as err:
         raise MadoopError(f"Failed executable test: {err}") from err
 
 
@@ -198,12 +206,17 @@ def map_single_chunk(exe, input_path, output_path, chunk):
                 check=True,
                 input=chunk,
                 stdout=outfile,
+                stderr=subprocess.PIPE
             )
         except subprocess.CalledProcessError as err:
             raise MadoopError(
                 f"Command returned non-zero: "
-                f"{exe} < {input_path} > {output_path}"
+                f"{exe} < {input_path} > {output_path}\n"
+                f"{err}"
+                f"\n{err.stderr.decode()}" if err.stderr else ""
             ) from err
+        except OSError as err:
+            raise MadoopError(f"Command returned non-zero: {err}") from err
 
 
 def map_stage(exe, input_dir, output_dir):
@@ -300,6 +313,7 @@ def partition_keys_custom(
             [partitioner, str(num_reducers)],
             stdin=stack.enter_context(inpath.open()),
             stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
             text=True,
         ))
         for line, partition in zip(
@@ -326,8 +340,10 @@ def partition_keys_custom(
 
         return_code = process.wait()
         if return_code:
+            stderr_output = process.stderr.read()
             raise MadoopError(
-                f"Partition executable returned non-zero: {str(partitioner)}"
+                f"Partition executable returned non-zero: {str(partitioner)}\n"
+                f"{stderr_output}"
             )
 
 
@@ -419,12 +435,17 @@ def reduce_single_file(exe, input_path, output_path):
                 check=True,
                 stdin=infile,
                 stdout=outfile,
+                stderr=subprocess.PIPE
             )
         except subprocess.CalledProcessError as err:
             raise MadoopError(
                 f"Command returned non-zero: "
-                f"{exe} < {input_path} > {output_path}"
+                f"{exe} < {input_path} > {output_path}\n"
+                f"{err}"
+                f"\n{err.stderr.decode()}" if err.stderr else ""
             ) from err
+        except OSError as err:
+            raise MadoopError(f"Command returned non-zero: {err}") from err
 
 
 def reduce_stage(exe, input_dir, output_dir):

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "madoop"
-version = "1.3.0"
+version = "1.3.1"
 description="A light weight MapReduce framework for education."
 license = {file = "LICENSE"}
 authors = [

tests/test_api.py

@@ -208,3 +208,64 @@ def test_input_path_spaces(tmpdir):
         TESTDATA_DIR/"word_count/correct/output",
         tmpdir/"output",
     )
+
+
+def test_map_exe_error_msg(tmpdir):
+    """Map exe returns non-zero with stderr output should produce an
+    error message and forward the stderr output.
+    """
+    with tmpdir.as_cwd(), pytest.raises(
+        madoop.MadoopError,
+        match="Map error message to stderr"
+    ):
+        madoop.mapreduce(
+            input_path=TESTDATA_DIR/"word_count/input",
+            output_dir="output",
+            map_exe=TESTDATA_DIR/"word_count/map_error_msg.py",
+            reduce_exe=TESTDATA_DIR/"word_count/reduce.py",
+            num_reducers=4,
+            partitioner=None,
+        )
+
+    with tmpdir.as_cwd(), pytest.raises(
+        madoop.MadoopError,
+        match="Map error message to stderr"
+    ):
+        map_stage(
+            exe=TESTDATA_DIR/"word_count/map_error_msg.py",
+            input_dir=TESTDATA_DIR/"word_count/input",
+            output_dir=Path(tmpdir),
+        )
+
+
+def test_partition_exe_error_msg(tmpdir):
+    """Partition exe returns non-zero with stderr output should produce an
+    error message and forward the stderr output.
+    """
+    with tmpdir.as_cwd(), pytest.raises(
+        madoop.MadoopError,
+        match="Partition error message to stderr"
+    ):
+        madoop.mapreduce(
+            input_path=TESTDATA_DIR/"word_count/input",
+            output_dir="output",
+            map_exe=TESTDATA_DIR/"word_count/map.py",
+            reduce_exe=TESTDATA_DIR/"word_count/reduce.py",
+            num_reducers=4,
+            partitioner=TESTDATA_DIR/"word_count/partition_error_msg.py",
+        )
+
+
+def test_reduce_exe_error_msg(tmpdir):
+    """Reduce exe returns non-zero with stderr output should produce an
+    error message and forward the stderr output.
+    """
+    with tmpdir.as_cwd(), pytest.raises(
+        madoop.MadoopError,
+        match="Reduce error message to stderr"
+    ):
+        reduce_stage(
+            exe=TESTDATA_DIR/"word_count/reduce_error_msg.py",
+            input_dir=TESTDATA_DIR/"word_count/input",
+            output_dir=Path(tmpdir),
+        )

tests/testdata/word_count/map_error_msg.py

@@ -0,0 +1,11 @@
+#!/usr/bin/env python3
+"""Invalid map executable returns non-zero with an error message."""
+
+import sys
+
+
+# Avoid error on executable check which has an empty string input
+input_lines_n = sum(1 for _ in sys.stdin)
+if input_lines_n > 1:
+    sys.stderr.write("Map error message to stderr\n")
+    sys.exit(1)

tests/testdata/word_count/partition_error_msg.py

@@ -0,0 +1,11 @@
+#!/usr/bin/env python3
+"""Invalid partition executable returns non-zero with an error message."""
+
+import sys
+
+# Avoid error on executable check which has an empty string input
+input_lines_n = sum(1 for _ in sys.stdin)
+if input_lines_n > 1:
+    sys.stderr.write("Partition error message to stdout\n")
+    sys.stderr.write("Partition error message to stderr\n")
+    sys.exit(1)

tests/testdata/word_count/reduce_error_msg.py

@@ -0,0 +1,11 @@
+#!/usr/bin/env python3
+"""Invalid reduce executable returns non-zero with an error message."""
+
+import sys
+
+
+# Avoid error on executable check which has an empty string input
+input_lines_n = sum(1 for _ in sys.stdin)
+if input_lines_n > 1:
+    sys.stderr.write("Reduce error message to stderr\n")
+    sys.exit(1)