python - 使用 Pyspark 进行单元测试 : unclosed socket warnings

Question

我想用 PySpark 进行单元测试。测试本身有效，但是对于我得到的每个测试

• ResourceWarning: unclosed <socket.socket [...]>和
• ResourceWarning: unclosed file <_io.BufferedWriter [...]>警告和
• DeprecationWarning关于 invalid escape sequence秒。

我想了解为什么/如何解决这个问题，以免这些警告弄乱我的单元测试输出。

这是一个 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')

如果我用 python pyspark_unittesting.py 调用它输出是:

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

如果我用 python -m unittest pyspark_unittesting 调用它然而，输出是:

/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>

编辑2018-03-29

关于@acue 的回答，我尝试使用 subprocess.Popen 调用脚本- 非常像在 unittest 中完成的模块:

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'                                                                                ']

没有出现资源警告...

Answer 1

如果您在测试的 setUp 方法中添加忽略它们的指令，这些警告将消失:

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)

现在运行 python3 -m unittest pyspark_unittesting.py 输出:

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