我想使用PySpark进行单元测试。测试本身可以正常运行,但是每个测试都会产生以下警告:
以下是一个最小化工作示例:
如果我使用
如果我使用
资源警告未显示...
ResourceWarning: unclosed <socket.socket [...]>
ResourceWarning: unclosed file <_io.BufferedWriter [...]>
DeprecationWarning
有关无效转义序列
的警告。
以下是一个最小化工作示例:
# 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年3月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' ']
资源警告未显示...