python - Pyspark 数据帧操作的单元测试用例

Question

我在python中编写了一些带有sql上下文的代码，即pyspark，通过将它们转换为pyspark数据帧来对csv执行一些操作（df操作，如预处理、重命名列名、创建新列并将它们附加到相同的数据帧等） . 我希望为它编写单元测试用例。我不知道在数据帧上编写单元测试用例。谁能帮助我如何在 pyspark 中的数据帧上编写单元测试用例？或者给我一些数据框测试用例的来源？

Answer 1

数据框与 pyspark 领域的其他任何东西没有什么不同。您可以从查看spark-testing-base 的 Python 部分开始。有几个有趣的项目有数据框测试，所以你可以开始看看他们是怎么做的：Sparkling Pandas就是其中一个，这里是另一个例子。还有find-spark可以帮助定位您的 spark 可执行上下文。但基本思想是在开始测试之前正确设置路径：

def add_pyspark_path():
    """
    Add PySpark to the PYTHONPATH
    Thanks go to this project: https://github.com/holdenk/sparklingpandas
    """
    import sys
    import os
    try:
        sys.path.append(os.path.join(os.environ['SPARK_HOME'], "python"))
        sys.path.append(os.path.join(os.environ['SPARK_HOME'],
            "python","lib","py4j-0.9-src.zip"))
    except KeyError:
        print "SPARK_HOME not set"
        sys.exit(1)

add_pyspark_path() # Now we can import pyspark

通常你会有一个基本的测试用例类：

import logging

from pyspark import SparkContext
from pyspark import SparkConf
from pyspark.sql import SQLContext, HiveContext

def quiet_py4j():
    """ turn down spark logging for the test context """
    logger = logging.getLogger('py4j')
    logger.setLevel(logging.WARN)

class SparkTestCase(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        quiet_py4j()

        # Setup a new spark context for each test
        conf = SparkConf()
        conf.set("spark.executor.memory","1g")
        conf.set("spark.cores.max", "1")
        #conf.set("spark.master", "spark://192.168.1.2:7077")
        conf.set("spark.app.name", "nosetest")
        cls.sc = SparkContext(conf=conf)
        cls.sqlContext = HiveContext(cls.sc)

    @classmethod
    def tearDownClass(cls):
        cls.sc.stop()