blocks|key|4452588|text|我使用pytest，它允许测试夹具，因此您可以实例化pyspark上下文，并将其注入到所有需要它的测试中。一些类似的东西|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|4452589|@pytest.fixture(scope="session",
++++++++++++++++params=[pytest.mark.spark_local('local'),
++++++++++++++++++++++++pytest.mark.spark_yarn('yarn')])
def+spark_context(request):
++++if+request.param+==+'local':
++++++++conf+=+(SparkConf()
++++++++++++++++.setMaster("local[2]")
++++++++++++++++.setAppName("pytest-pyspark-local-testing")
++++++++++++++++)
++++elif+request.param+==+'yarn':
++++++++conf+=+(SparkConf()
++++++++++++++++.setMaster("yarn-client")
++++++++++++++++.setAppName("pytest-pyspark-yarn-testing")
++++++++++++++++.set("spark.executor.memory",+"1g")
++++++++++++++++.set("spark.executor.instances",+2)
++++++++++++++++)
++++request.addfinalizer(lambda:+sc.stop())

++++sc+=+SparkContext(conf=conf)
++++return+sc

def+my_test_that_requires_sc(spark_context):
++++assert+spark_context.textFile('/path/to/a/file').count()+==+10|code-block|syntax|javascript|4452590|然后，您可以通过调用py.test+-m+spark_local在本地模式下运行测试，或者在带有py.test+-m+spark_yarn的YARN中运行测试。这对我来说工作得很好。|4452591|entityMap^0|3|6|0|0|A|M|1C|L|0^^$0|@$1|2|3|4|5|6|7|O|8|@$9|P|A|Q|B|C]]|D|@]|E|$]]|$1|F|3|G|5|H|7|R|8|@]|D|@]|E|$I|J]]|$1|K|3|L|5|6|7|S|8|@$9|T|A|U|B|C]|$9|V|A|W|B|C]]|D|@]|E|$]]|$1|M|3|-4|5|6|7|X|8|@]|D|@]|E|$]]]|N|$]]

I use <code>pytest</code>, which allows test fixtures so you can instantiate a pyspark context and inject it into all of your tests that require it. Something along the lines of 

<pre><code>@pytest.fixture(scope="session",
 params=[pytest.mark.spark_local('local'),
 pytest.mark.spark_yarn('yarn')])
def spark_context(request):
 if request.param == 'local':
 conf = (SparkConf()
 .setMaster("local[2]")
 .setAppName("pytest-pyspark-local-testing")
 )
 elif request.param == 'yarn':
 conf = (SparkConf()
 .setMaster("yarn-client")
 .setAppName("pytest-pyspark-yarn-testing")
 .set("spark.executor.memory", "1g")
 .set("spark.executor.instances", 2)
 )
 request.addfinalizer(lambda: sc.stop())

 sc = SparkContext(conf=conf)
 return sc

def my_test_that_requires_sc(spark_context):
 assert spark_context.textFile('/path/to/a/file').count() == 10
</code></pre>

Then you can run the tests in local mode by calling <code>py.test -m spark_local</code> or in YARN with <code>py.test -m spark_yarn</code>. This has worked pretty well for me.

blocks|key|4452432|text|我也推荐使用py.test。py.test使得创建可重用的SparkContext测试夹具并使用它来编写简洁的测试函数变得很容易。您还可以专门化fixture(例如，创建一个StreamingContext+)，并在测试中使用其中的一个或多个。|type|unstyled|depth|inlineStyleRanges|entityRanges|data|4452433|我在Medium上写了一篇关于这个话题的博客文章：|4452434|https://engblog.nextdoor.com/unit-testing-apache-spark-with-py-test-3b8970dc013b|offset|length|4452435|以下是帖子中的一段代码：|4452436|pytestmark+=+pytest.mark.usefixtures("spark_context")
def+test_do_word_counts(spark_context):
++++"""+test+word+couting
++++Args:
+++++++spark_context:+test+fixture+SparkContext
++++"""
++++test_input+=+[
++++++++'+hello+spark+',
++++++++'+hello+again+spark+spark'
++++]

++++input_rdd+=+spark_context.parallelize(test_input,+1)
++++results+=+wordcount.do_word_counts(input_rdd)

++++expected_results+=+{'hello':2,+'spark':3,+'again':1}++
++++assert+results+==+expected_results|code-block|syntax|javascript|4452437|entityMap|0|LINK|mutability|MUTABLE|url^0|0|0|0|28|0|0|0|0^^$0|@$1|2|3|4|5|6|7|V|8|@]|9|@]|A|$]]|$1|B|3|C|5|6|7|W|8|@]|9|@]|A|$]]|$1|D|3|E|5|6|7|X|8|@]|9|@$F|Y|G|Z|1|10]]|A|$]]|$1|H|3|I|5|6|7|11|8|@]|9|@]|A|$]]|$1|J|3|K|5|L|7|12|8|@]|9|@]|A|$M|N]]|$1|O|3|-4|5|6|7|13|8|@]|9|@]|A|$]]]|P|$Q|$5|R|S|T|A|$U|E]]]]

I'd recommend using py.test as well. py.test makes it easy to create re-usable SparkContext test fixtures and use it to write concise test functions. You can also specialize fixtures (to create a StreamingContext for example) and use one or more of them in your tests. 

I wrote a blog post on Medium on this topic:

<a href="https://engblog.nextdoor.com/unit-testing-apache-spark-with-py-test-3b8970dc013b">https://engblog.nextdoor.com/unit-testing-apache-spark-with-py-test-3b8970dc013b</a>

Here is a snippet from the post:

<pre><code>pytestmark = pytest.mark.usefixtures("spark_context")
def test_do_word_counts(spark_context):
 """ test word couting
 Args:
 spark_context: test fixture SparkContext
 """
 test_input = [
 ' hello spark ',
 ' hello again spark spark'
 ]

 input_rdd = spark_context.parallelize(test_input, 1)
 results = wordcount.do_word_counts(input_rdd)

 expected_results = {'hello':2, 'spark':3, 'again':1} 
 assert results == expected_results
</code></pre>

blocks|key|4452626|text|假设您已经安装了pyspark，您可以使用下面的类在unittest中对其进行unitTest|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|4452627|import+unittest
import+pyspark


class+PySparkTestCase(unittest.TestCase):

++++@classmethod
++++def+setUpClass(cls):
++++++++conf+=+pyspark.SparkConf().setMaster("local[2]").setAppName("testing")
++++++++cls.sc+=+pyspark.SparkContext(conf=conf)
++++++++cls.spark+=+pyspark.SQLContext(cls.sc)

++++@classmethod
++++def+tearDownClass(cls):
++++++++cls.sc.stop()|code-block|syntax|javascript|4452628|示例：|4452629|class+SimpleTestCase(PySparkTestCase):

++++def+test_with_rdd(self):
++++++++test_input+=+[
++++++++++++'+hello+spark+',
++++++++++++'+hello+again+spark+spark'
++++++++]

++++++++input_rdd+=+self.sc.parallelize(test_input,+1)

++++++++from+operator+import+add

++++++++results+=+input_rdd.flatMap(lambda+x:+x.split()).map(lambda+x:+(x,+1)).reduceByKey(add).collect()
++++++++self.assertEqual(results,+[('hello',+2),+('spark',+3),+('again',+1)])

++++def+test_with_df(self):
++++++++df+=+self.spark.createDataFrame(data=[[1,+'a'],+[2,+'b']],+
++++++++++++++++++++++++++++++++++++++++schema=['c1',+'c2'])
++++++++self.assertEqual(df.count(),+2)|4452630|请注意，这将为每个类创建一个上下文。使用setUp而不是setUpClass来获取每个测试的上下文。这通常会增加执行测试的大量开销时间，因为创建新的spark上下文目前代价很高。|4452631|entityMap^0|8|7|Q|8|0|0|0|0|K|5|S|A|0^^$0|@$1|2|3|4|5|6|7|S|8|@$9|T|A|U|B|C]|$9|V|A|W|B|C]]|D|@]|E|$]]|$1|F|3|G|5|H|7|X|8|@]|D|@]|E|$I|J]]|$1|K|3|L|5|6|7|Y|8|@]|D|@]|E|$]]|$1|M|3|N|5|H|7|Z|8|@]|D|@]|E|$I|J]]|$1|O|3|P|5|6|7|10|8|@$9|11|A|12|B|C]|$9|13|A|14|B|C]]|D|@]|E|$]]|$1|Q|3|-4|5|6|7|15|8|@]|D|@]|E|$]]]|R|$]]

Assuming you have <code>pyspark</code> installed, you can use the class below for unitTest it in <code>unittest</code>:

<pre><code>import unittest
import pyspark


class PySparkTestCase(unittest.TestCase):

 @classmethod
 def setUpClass(cls):
 conf = pyspark.SparkConf().setMaster("local[2]").setAppName("testing")
 cls.sc = pyspark.SparkContext(conf=conf)
 cls.spark = pyspark.SQLContext(cls.sc)

 @classmethod
 def tearDownClass(cls):
 cls.sc.stop()
</code></pre>

Example:

<pre><code>class SimpleTestCase(PySparkTestCase):

 def test_with_rdd(self):
 test_input = [
 ' hello spark ',
 ' hello again spark spark'
 ]

 input_rdd = self.sc.parallelize(test_input, 1)

 from operator import add

 results = input_rdd.flatMap(lambda x: x.split()).map(lambda x: (x, 1)).reduceByKey(add).collect()
 self.assertEqual(results, [('hello', 2), ('spark', 3), ('again', 1)])

 def test_with_df(self):
 df = self.spark.createDataFrame(data=[[1, 'a'], [2, 'b']], 
 schema=['c1', 'c2'])
 self.assertEqual(df.count(), 2)
</code></pre>

Note that this creates a context per class. Use <code>setUp</code> instead of <code>setUpClass</code> to get a context per test. This typically adds a lot of overhead time on the execution of the tests, as creating a new spark context is currently expensive.

blocks|key|4452487|text|pyspark有unittest模块，使用方法如下|type|unstyled|depth|inlineStyleRanges|entityRanges|data|4452488|from+pyspark.tests+import+ReusedPySparkTestCase+as+PySparkTestCase

class+MySparkTests(PySparkTestCase):
++++def+spark_session(self):
++++++++return+pyspark.SQLContext(self.sc)

++++def+createMockDataFrame(self):
+++++++++self.spark_session().createDataFrame(
++++++++++++[
++++++++++++++++("t1",+"t2"),
++++++++++++++++("t1",+"t2"),
++++++++++++++++("t1",+"t2"),
++++++++++++],
++++++++++++['col1',+'col2']
++++++++)|code-block|syntax|javascript|4452489|entityMap^0|0|0^^$0|@$1|2|3|4|5|6|7|I|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|J|8|@]|9|@]|A|$E|F]]|$1|G|3|-4|5|6|7|K|8|@]|9|@]|A|$]]]|H|$]]

pyspark has unittest module which can be used as below
<pre class="lang-py prettyprint-override"><code>from pyspark.tests import ReusedPySparkTestCase as PySparkTestCase

class MySparkTests(PySparkTestCase):
 def spark_session(self):
 return pyspark.SQLContext(self.sc)

 def createMockDataFrame(self):
 self.spark_session().createDataFrame(
 [
 (&quot;t1&quot;, &quot;t2&quot;),
 (&quot;t1&quot;, &quot;t2&quot;),
 (&quot;t1&quot;, &quot;t2&quot;),
 ],
 ['col1', 'col2']
 )
</code></pre>

My current Java/Spark Unit Test approach works (detailed <a href="https://stackoverflow.com/a/32213314/2596363">here</a>) by instantiating a SparkContext using "local" and running unit tests using JUnit.

The code has to be organized to do I/O in one function and then call another with multiple RDDs.

This works great. I have a highly tested data transformation written in Java + Spark.

Can I do the same with Python?

How would I run Spark unit tests with Python?

How do I unit test PySpark programs?

翻译质量差，导致语言生硬或混乱。

没有提供实际的解决方法或示例。

解答不清晰，无法理解或解决问题。

页面排版不美观，阅读体验差。

文章

问答

视频

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

我目前的Java/Spark单元测试方法(详细的)是通过使用“本地”实例化SparkContext并使用JUnit运行单元测试来实现的。必须组织代码，以便在一个函数中执行I/O，然后使用多个RDD调用另一个函数。这很好用。我有一个用Java + Spark编写的经过高度测试的数据转换。我能用Python做同样的事情吗？...

问如何对PySpark程序进行单元测试？
EN

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问如何对PySpark程序进行单元测试？EN

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问如何对PySpark程序进行单元测试？
EN