Unittesting with Pyspark: unclosed socket warnings

Question

I want to do unittesting with PySpark. The tests itself work, however for each test I get

• ResourceWarning: unclosed <socket.socket [...]> and
• ResourceWarning: unclosed file <_io.BufferedWriter [...]> warnings and
• DeprecationWarnings regarding invalid escape sequences.

I'd like to understand why / how to solve this to not clutter my unittest output with these warnings.

Here is a MWE:

# filename: pyspark_unittesting.py
# -*- coding: utf-8 -*-

import unittest


def insert_and_collect(val_in):
    from pyspark.sql import SparkSession
    with SparkSession.builder.getOrCreate() as spark:
        col = 'column_x'
        df = spark.createDataFrame([(val_in,)], [col])

        print('one')
        print(df.count())
        print('two')
        collected = df.collect()
        print('three')
        return collected[0][col]


class MyTest(unittest.TestCase):
    def test(self):
        val = 1
        self.assertEqual(insert_and_collect(val), val)
        print('four')


if __name__ == '__main__':
    val = 1
    print('inserted and collected is equal to original: {}'
          .format(insert_and_collect(val) == val))
    print('five')

If I call this with python pyspark_unittesting.py the output is:

Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
one
1  
two
three
inserted and collected is equal to original: True
five

If I call this with python -m unittest pyspark_unittesting however, the output is:

/opt/spark/current/python/lib/py4j-0.10.4-src.zip/py4j/java_gateway.py:1890: DeprecationWarning: invalid escape sequence \*
/opt/spark/current/python/lib/py4j-0.10.4-src.zip/py4j/java_gateway.py:1890: DeprecationWarning: invalid escape sequence \*
/opt/spark/current/python/lib/pyspark.zip/pyspark/sql/readwriter.py:398: DeprecationWarning: invalid escape sequence \`
/opt/spark/current/python/lib/pyspark.zip/pyspark/sql/readwriter.py:759: DeprecationWarning: invalid escape sequence \`
/opt/spark/current/python/lib/pyspark.zip/pyspark/sql/readwriter.py:398: DeprecationWarning: invalid escape sequence \`
/opt/spark/current/python/lib/pyspark.zip/pyspark/sql/readwriter.py:759: DeprecationWarning: invalid escape sequence \`
/opt/spark/current/python/lib/pyspark.zip/pyspark/sql/streaming.py:618: DeprecationWarning: invalid escape sequence \`
/opt/spark/current/python/lib/pyspark.zip/pyspark/sql/streaming.py:618: DeprecationWarning: invalid escape sequence \`
/opt/spark/current/python/lib/pyspark.zip/pyspark/sql/functions.py:1519: DeprecationWarning: invalid escape sequence \d
/opt/spark/current/python/lib/pyspark.zip/pyspark/sql/functions.py:1519: DeprecationWarning: invalid escape sequence \d
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
/usr/lib/python3.6/subprocess.py:766: ResourceWarning: subprocess 10219 is still running
  ResourceWarning, source=self)
/usr/lib/python3.6/importlib/_bootstrap.py:219: ImportWarning: can't resolve package from __spec__ or __package__, falling back on __name__ and __path__
  return f(*args, **kwds)
one
1                                                                               
two
/usr/lib/python3.6/socket.py:657: ResourceWarning: unclosed <socket.socket fd=7, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=6, laddr=('127.0.0.1', 49330), raddr=('127.0.0.1', 44169)>
  self._sock = None
three
four
.
----------------------------------------------------------------------
Ran 1 test in 7.394s

OK
sys:1: ResourceWarning: unclosed file <_io.BufferedWriter name=5>

Edit 2018-03-29

Regarding @acue's answer, I tried out calling the script using subprocess.Popen - very much like it is done within the unittest module:

In [1]: import pathlib
      : import subprocess
      : import sys
      : 
      : here = pathlib.Path('.').absolute()
      : args = [sys.executable, str(here / 'pyspark_unittesting.py')]
      : opts = dict(stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd='/tmp')
      : p = subprocess.Popen(args, **opts)
      : out, err = [b.splitlines() for b in p.communicate()]
      : print(out)
      : print(err)
      : 
      : 
[b'one',
 b'1',
 b'two',
 b'three',
 b'inserted and collected is equal to original: True',
 b'five']

[b"Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties",
 b'Setting default log level to "WARN".',
 b'To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).',
 b'',
 b'[Stage 0:>                                                          (0 + 0) / 8]',
 b'[Stage 0:>                                                          (0 + 8) / 8]',
 b'                                                                                ']

The Resource Warnings do not appear...

Answer 1

These warnings will go away if you add the instruction to ignore them in the setUp method of the test:

import unittest
import warnings


class MyTest(unittest.TestCase):
    def test(self):
        val = 1
        self.assertEqual(insert_and_collect(val), val)
        print('four')

    def setUp(self):
        warnings.filterwarnings("ignore", category=ResourceWarning)
        warnings.filterwarnings("ignore", category=DeprecationWarning)

Now running with python3 -m unittest pyspark_unittesting.py outputs:

Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
one
1                                                                               
two
three
four
.
----------------------------------------------------------------------
Ran 1 test in 11.124s

OK