开始
用户指南
管理指南
参考指南
资源
2023年1月13日更新
给我们反馈
你可以使用单元测试帮助提高笔记本代码的质量和一致性。单元测试是一种早期和经常测试自包含代码单元(如函数)的方法。这可以帮助您更快地发现代码中的问题,更快地发现关于代码的错误假设,并简化您的整体编码工作。
这篇文章是基本的介绍单元测试与功能。的高级概念,如单元测试类和接口,以及的使用存根,模拟,测试,但在对笔记本进行单元测试时也受支持,这超出了本文的范围。本文也不涉及其他类型的测试方法,例如集成测试,系统测试,验收测试,或非功能性测试方法包括性能测试orgydF4y2Ba可用性测试.
本文演示了以下内容:
如何组织函数及其单元测试。
如何在Python、R、Scala中编写函数,以及在SQL中编写用户定义的函数,这些函数都是为单元测试而精心设计的。
如何从Python、R、Scala和SQL笔记本中调用这些函数。
如何使用流行的测试框架在Python、R和Scala中编写单元测试pytest对于Python,testthat对于R,并且——scalateScala。还有如何编写单元测试SQL用户定义函数(SQL udf)的SQL。
如何从Python、R、Scala和SQL笔记本运行这些单元测试。
有几种常用的方法可以用笔记本组织函数及其单元测试。每种方法都有其优点和挑战。
对于Python、R和Scala笔记本,常见的方法包括:
将函数及其单元测试存储在笔记本电脑之外。.
优点:你可以在笔记本内和笔记本外调用这些函数。测试框架更适合在笔记本电脑之外运行测试。
挑战:Scala笔记本不支持这种方法。这种方法需要砖回购.这种方法还增加了要跟踪和维护的文件数量。
将函数存储在一个笔记本中,将它们的单元测试存储在另一个笔记本中。.
优点:这些函数更容易在笔记本上重用。
挑战:需要跟踪和维护的笔记本数量增加了。这些功能不能在笔记本电脑之外使用。这些功能在笔记本电脑之外也更难测试。
将函数及其单元测试存储在同一个笔记本中。.
优点:函数及其单元测试存储在单个笔记本中,以便于跟踪和维护。
挑战:在笔记本电脑之间重用这些函数可能更加困难。这些功能不能在笔记本电脑之外使用。这些功能在笔记本电脑之外也更难测试。
对于Python和R笔记本电脑,Databricks建议将函数及其单元测试存储在笔记本电脑之外。对于Scala笔记本,Databricks建议将函数放在一个笔记本中,将单元测试放在另一个笔记本中。
对于SQL笔记本,Databricks建议将函数作为SQL用户定义函数(SQL udf)存储在模式(也称为数据库)中。然后,您可以从SQL笔记本调用这些SQL udf及其单元测试。
本节描述了一组简单的示例函数,它们确定以下内容:
数据库中是否存在表。
表中是否存在列。
对于列中的值,一列中存在多少行。
这些函数都很简单,这样您就可以专注于本文中的单元测试细节,而不是函数本身。
为了获得最佳的单元测试结果,函数应该返回单个可预测的结果,并且是单个数据类型。例如,为了检查某个东西是否存在,函数应该返回一个布尔值true或false。为了返回已存在的行数,函数应该返回一个非负的整数。在第一个例子中,如果某物不存在,它不应该返回false;如果某物存在,它也不应该返回事物本身。同样,对于第二个示例,它既不返回已存在的行数,也不返回不存在行的false。
您可以将这些函数添加到现有的Databricks工作空间中,使用Python、R、Scala或SQL。
下面的代码假设您有设置Databricks回购,增加了一个回购,并在Databricks工作区中打开repo。
创建文件命名myfunctions.py在repo中,将以下内容添加到文件中。本文中的其他示例希望命名此文件myfunctions.py.您可以为自己的文件使用不同的名称。
myfunctions.py
进口pyspark从pyspark.sql进口SparkSession从pyspark.sql.functions进口上校#因为这个文件不是一个Databricks笔记本,你必须创建一个Spark会话。砖的笔记本默认为您创建一个Spark会话。火花=SparkSession.构建器\.浏览器名称(“诚信测试”)\.getOrCreate()#指定的表在指定的数据库中是否存在?deftableExists(的表,dbName):返回火花.目录.tableExists(f"{dbName}.{的表}")#指定的列是否存在于给定的数据框架中?defcolumnExists(dataFrame,columnName):如果columnName在dataFrame.列:返回真正的其他的:返回假#在指定列中指定值有多少行#在给定的DataFrame?defnumRowsInColumnForValue(dataFrame,columnName,columnValue):df=dataFrame.过滤器(上校(columnName)= =columnValue)返回df.数()
创建文件命名myfunctions.r在repo中,将以下内容添加到文件中。本文中的其他示例希望命名此文件myfunctions.r.您可以为自己的文件使用不同的名称。
myfunctions.r
图书馆(SparkR)#指定的表在指定的数据库中是否存在?table_exists<-函数(table_name,db_name){tableExists(粘贴(db_name,“。”,table_name,9月=""))}#指定的列是否存在于给定的数据框架中?column_exists<-函数(dataframe,column_name){column_name%在%colnames(dataframe)}#在指定列中指定值有多少行#在给定的DataFrame?num_rows_in_column_for_value<-函数(dataframe,column_name,column_value){df=过滤器(dataframe,dataframe[[column_name]]= =column_value)数(df)}
创建一个Scala的笔记本命名myfunction附有以下内容。本文中的其他示例都希望命名这个笔记本myfunction.您可以为自己的笔记本使用不同的名称。
myfunction
进口org.apache.火花.sql.DataFrame进口org.apache.火花.sql.功能.上校//指定的表是否存在于指定的数据库中?deftableExists(的表:字符串,dbName:字符串):布尔={返回火花.目录.tableExists(dbName+“。”+的表)}//指定的列是否存在于给定的DataFrame中?defcolumnExists(dataFrame:DataFrame,columnName:字符串):布尔={瓦尔nameOfColumn=零为(nameOfColumn<-dataFrame.列){如果(nameOfColumn= =columnName){返回真正的}}返回假}//在指定的列中指定的值有多少行//在给定的DataFrame?defnumRowsInColumnForValue(dataFrame:DataFrame,columnName:字符串,columnValue:字符串):长={瓦尔df=dataFrame.过滤器(上校(columnName)= = =columnValue)返回df.数()}
下面的代码假设您拥有第三方示例数据集钻石在一个名为默认的在一个名为主要可以从Databricks工作区访问。如果要使用的编目或模式具有不同的名称,则更改以下一个或两个名称使用语句匹配。
默认的
主要
使用
创建一个SQL的笔记本并将以下内容添加到这个新笔记本中。然后附加笔记本到集群和运行将以下SQL udf添加到指定的目录和模式。
请注意
SQL udftable_exists而且column_exists工作只与统一目录。SQL UDF支持统一目录公共预览.
table_exists
column_exists
使用目录主要;使用模式默认的;创建或取代函数table_exists(catalog_name字符串,db_name字符串,table_name字符串)返回布尔返回如果((选择数(*)从系统.information_schema.表在哪里table_catalog=table_exists.catalog_name和table_schema=table_exists.db_name和table_name=table_exists.table_name)>0,真正的,假);创建或取代函数column_exists(catalog_name字符串,db_name字符串,table_name字符串,column_name字符串)返回布尔返回如果((选择数(*)从系统.information_schema.列在哪里table_catalog=column_exists.catalog_name和table_schema=column_exists.db_name和table_name=column_exists.table_name和column_name=column_exists.column_name)>0,真正的,假);创建或取代函数num_rows_for_clarity_in_diamonds(clarity_value字符串)返回长整型数字返回选择数(*)从主要.默认的.钻石在哪里清晰=clarity_value
本节描述调用上述函数的代码。例如,您可以使用这些函数来计算表中指定列中存在指定值的行数。但是,在继续之前,您可能希望检查表是否实际存在,以及列是否实际存在于该表中。下面的代码检查这些条件。
如果将上一节中的函数添加到Databricks工作空间,则可以从工作空间调用这些函数,如下所示。
创建一个Python笔记本和之前的文件夹一样myfunctions.py在你的回购文件中,并将以下内容添加到笔记本中。根据需要更改表名、模式(数据库)名、列名和列值的变量值。然后附加笔记本到集群和运行看一下笔记本的结果。
从myfunction进口*的表=“钻石”dbName=“默认”columnName=“清晰”columnValue=“VVS2”如果表在指定的数据库中存在…如果tableExists(的表,dbName):df=火花.sql(fselect * from{dbName}.{的表}")#指定的列存在于那个表中…如果columnExists(df,columnName):#然后报告该列中指定值的行数。numRows=numRowsInColumnForValue(df,columnName,columnValue)打印(f“有{numRows}排在几排{的表}“哪里”{columnName}'等号'{columnValue}’。”)其他的:打印(f“列{columnName}'表中不存在'{的表}' in schema (database) '{dbName}’。”)其他的:打印(f“表{的表}'模式(数据库)中不存在'{dbName}’。”)
创建一个R笔记本和之前的文件夹一样myfunctions.r在你的回购文件中,并将以下内容添加到笔记本中。根据需要更改表名、模式(数据库)名、列名和列值的变量值。然后附加笔记本到集群和运行看一下笔记本的结果。
图书馆(SparkR)源(“myfunctions.r”)table_name<-“钻石”db_name<-“默认”column_name<-“清晰”column_value<-“VVS2”如果表在指定的数据库中存在…如果(table_exists(table_name,db_name)){df=sql(粘贴(“select * from”,db_name,“。”,table_name,9月=""))#指定的列存在于那个表中…如果(column_exists(df,column_name)){#然后报告该列中指定值的行数。num_rows=num_rows_in_column_for_value(df,column_name,column_value)打印(粘贴(“有”,num_rows,"表中的行",table_name,“在哪里”,column_name,“等于”,column_value,”’。”,9月=""))}其他的{打印(粘贴(“列”,column_name,"表中不存在",table_name,在schema(数据库)中’”,db_name,”’。”,9月=""))}}其他的{打印(粘贴(“表”,table_name,"'在模式(数据库)中不存在’”,db_name,”’。”,9月=""))}
在前面的文件夹中创建另一个Scala笔记本myfunction并将以下内容添加到这个新笔记本中。
在这个新笔记本的第一个单元格中,添加以下代码,该代码调用运行%魔法。这魔法使内容myfunction笔记本可用于您的新笔记本。
%。/ myfunction
在这个新笔记本的第二个单元格中,添加以下代码。根据需要更改表名、模式(数据库)名、列名和列值的变量值。然后附加笔记本到集群和运行看一下笔记本的结果。
瓦尔的表=“钻石”瓦尔dbName=“默认”瓦尔columnName=“清晰”瓦尔columnValue=“VVS2”//如果表在指定的数据库中存在…如果(tableExists(的表,dbName)){瓦尔df=火花.sql(“select * from”+dbName+“。”+的表)//指定的列存在于该表中…如果(columnExists(df,columnName)){//然后报告该列中指定值的行数。瓦尔numRows=numRowsInColumnForValue(df,columnName,columnValue)println(“有”+numRows+“排在一起”+的表+“在哪里”+columnName+“等于”+columnValue+”’。”)}其他的{println(“列”+columnName+"表中不存在"+的表+"在数据库中"+dbName+”’。”)}}其他的{println(“表”+的表+"数据库中不存在"+dbName+”’。”)}
将以下代码添加到上一个笔记本中的新单元格或另一个笔记本中的单元格。如果需要,更改模式或目录名称以与您的相匹配,然后运行此单元格以查看结果。
选择情况下——如果表存在于指定的目录和模式中…当table_exists(“主要”,“默认”,“钻石”)然后——指定的列存在于那个表中…(选择情况下当column_exists(“主要”,“默认”,“钻石”,“清晰”)然后——然后报告该列中指定值的行数。printf(在表'main.default.diamonds'中有%d行,其中'clarity'等于'VVS2'。",num_rows_for_clarity_in_diamonds(“VVS2”))其他的printf(“表‘main.default.diamonds’中不存在列‘clarity’。”)结束)其他的printf(“表main.default.diamonds不存在。”)结束
本节描述了测试本文开头所描述的每个函数的代码。如果将来对函数进行任何更改,则可以使用单元测试来确定这些函数是否仍按预期工作。
如果您将本文开头的函数添加到Databricks工作空间中,那么您可以如下所示将这些函数的单元测试添加到您的工作空间中。
创建另一个名为test_myfunctions.py和之前的文件夹一样myfunctions.py文件在您的回购,并添加以下内容的文件。默认情况下,pytest查找. py名称以test_(或以_t)进行测试。同样,默认情况下,pytest在这些文件中查找名称以test_进行测试。
test_myfunctions.py
pytest
. py
test_
_t
一般来说,这是一种最佳实践不针对使用生产中数据的函数运行单元测试。这对于添加、删除或以其他方式更改数据的函数尤其重要。为了保护生产数据不被单元测试以意想不到的方式破坏,应该针对非生产数据运行单元测试。一种常见的方法是创建与生产数据尽可能接近的假数据。下面的代码示例为要运行的单元测试创建假数据。
进口pytest进口pyspark从myfunction进口*从pyspark.sql进口SparkSession从pyspark.sql.types进口StructType,StructField,IntegerType,FloatType,StringType的表=“钻石”dbName=“默认”columnName=“清晰”columnValue=“SI2”#因为这个文件不是一个Databricks笔记本,你必须创建一个Spark会话。砖的笔记本默认为您创建一个Spark会话。火花=SparkSession.构建器\.浏览器名称(“诚信测试”)\.getOrCreate()#为运行的单元测试创建假数据。#一般来说,不运行单元测试是一种最佳实践#针对在生产环境中使用数据的函数。模式=StructType([\StructField(“_c0”,IntegerType(),真正的),\StructField(“克拉”,FloatType(),真正的),\StructField(“切”,StringType(),真正的),\StructField(“颜色”,StringType(),真正的),\StructField(“清晰”,StringType(),真正的),\StructField(“深度”,FloatType(),真正的),\StructField(“表”,IntegerType(),真正的),\StructField(“价格”,IntegerType(),真正的),\StructField(“x”,FloatType(),真正的),\StructField(“y”,FloatType(),真正的),\StructField(“z”,FloatType(),真正的),\])数据=[(1,0.23,“理想”,“E”,“SI2”,61.5,55,326,3.95,3.98,2.43),\(2,0.21,“溢价”,“E”,“SI1”,59.8,61,326,3.89,3.84,2.31)]df=火花.createDataFrame(数据,模式)#表是否存在?deftest_tableExists():断言tableExists(的表,dbName)是真正的#列是否存在?deftest_columnExists():断言columnExists(df,columnName)是真正的在指定的列中至少有一行的值?deftest_numRowsInColumnForValue():断言numRowsInColumnForValue(df,columnName,columnValue)>0
创建另一个名为test_myfunctions.r和之前的文件夹一样myfunctions.r文件在您的回购,并添加以下内容的文件。默认情况下,testthat查找.r名称以测验进行测试。
test_myfunctions.r
testthat
.r
测验
图书馆(testthat)源(“myfunctions.r”)table_name<-“钻石”db_name<-“默认”column_name<-“清晰”column_value<-“SI2”#为运行的单元测试创建假数据。#一般来说,不运行单元测试是一种最佳实践#针对在生产环境中使用数据的函数。模式<-structType(structField(“_c0”,“整数”),structField(“克拉”,“浮动”),structField(“切”,“字符串”),structField(“颜色”,“字符串”),structField(“清晰”,“字符串”),structField(“深度”,“浮动”),structField(“表”,“整数”),structField(“价格”,“整数”),structField(“x”,“浮动”),structField(“y”,“浮动”),structField(“z”,“浮动”))数据<-列表(列表(作为.整数(1),0.23,“理想”,“E”,“SI2”,61.5,作为.整数(55),作为.整数(326),3.95,3.98,2.43),列表(作为.整数(2),0.21,“溢价”,“E”,“SI1”,59.8,作为.整数(61),作为.整数(326),3.89,3.84,2.31))df<-createDataFrame(数据,模式)#表是否存在?test_that(“桌子是存在的。”,{expect_true(table_exists(table_name,db_name))})#列是否存在?test_that(“列存在于表中。”,{expect_true(column_exists(df,column_name))})在指定的列中至少有一行的值?test_that(“查询结果中至少有一行。”,{expect_true(num_rows_in_column_for_value(df,column_name,column_value)>0)})
在新笔记本的第一个单元格中,添加以下代码,该代码调用运行%魔法。这魔法使内容myfunction笔记本可用于您的新笔记本。
运行%
在第二个单元格中,添加以下代码。此代码定义单元测试并指定如何运行它们。
进口org.——scalate._进口org.apache.火花.sql.类型.{StructType,StructField,IntegerType,FloatType,StringType}进口scala.集合.JavaConverters._类DataTests扩展AsyncFunSuite{瓦尔的表=“钻石”瓦尔dbName=“默认”瓦尔columnName=“清晰”瓦尔columnValue=“SI2”//为运行的单元测试创建假数据。//一般来说,不运行单元测试是最好的做法//针对在生产环境中处理数据的函数。瓦尔模式=StructType(数组(StructField(“_c0”,IntegerType),StructField(“克拉”,FloatType),StructField(“切”,StringType),StructField(“颜色”,StringType),StructField(“清晰”,StringType),StructField(“深度”,FloatType),StructField(“表”,IntegerType),StructField(“价格”,IntegerType),StructField(“x”,FloatType),StructField(“y”,FloatType),StructField(“z”,FloatType)))瓦尔数据=Seq(行(1,0.23,“理想”,“E”,“SI2”,61.5,55,326,3.95,3.98,2.43),行(2,0.21,“溢价”,“E”,“SI1”,59.8,61,326,3.89,3.84,2.31)).asJava瓦尔df=火花.createDataFrame(数据,模式)//表是否存在?测验("表存在"){断言(tableExists(的表,dbName)= =真正的)}//列是否存在?测验(“专栏存在”){断言(columnExists(df,columnName)= =真正的)}//在指定的列中至少有一行的值?测验("至少有一个匹配的行"){断言(numRowsInColumnForValue(df,columnName,columnValue)>0)}}nocolor.nodurations.nostacks.统计数据.运行(新DataTests)
此代码示例使用FunSuiteScalaTest的测试风格。有关其他可用的测试样式,请参见为项目选择测试样式.
FunSuite
在添加单元测试之前,您应该意识到,一般来说,最好的实践是不针对使用生产中数据的函数运行单元测试。这对于添加、删除或以其他方式更改数据的函数尤其重要。为了保护生产数据不被单元测试以意想不到的方式破坏,应该针对非生产数据运行单元测试。一种常见的方法是运行单元测试的观点而不是表格。
要创建视图,可以调用创建视图从上面的笔记本或单独的笔记本中的新单元中命令。下面的示例假设您有一个名为钻石在命名为默认的在一个名为主要.根据需要更改这些名称以匹配您自己的名称,然后只运行该单元格。
钻石
使用目录主要;使用模式默认的;创建视图view_diamonds作为选择*从钻石;
创建视图后,添加以下每一个视图选择语句到前一个笔记本中它自己的新单元格中,或到另一个笔记本中它自己的新单元格中。根据需要更改名称以匹配您自己的名称。
选择
选择如果(table_exists(“主要”,“默认”,“view_diamonds”),printf(PASS:表'main.default.view_diamonds'存在。),printf(失败:表'main.default.view_diamonds'不存在。));选择如果(column_exists(“主要”,“默认”,“view_diamonds”,“清晰”),printf(PASS: clear列存在于表main.default.view_diamonds中。),printf(失败:表main.default.view_diamonds中不存在列clear。));选择如果(num_rows_for_clarity_in_diamonds(“VVS2”)>0,printf(PASS:表'main.default.view_diamonds'至少有一行,其中列'clarity'等于'VVS2'。),printf(失败:表'main.default.view_diamonds'没有至少一行列'clarity'等于'VVS2'。));
本节描述如何运行在前一节中编写的单元测试。运行单元测试时,将得到显示哪些单元测试通过和失败的结果。
如果将上一节中的单元测试添加到Databricks工作空间,则可以从工作空间运行这些单元测试。您也可以运行这些单元测试手动orgydF4y2Ba按照时间表.
在上面的文件夹中创建一个Python笔记本test_myfunctions.py文件在您的回购,并添加以下内容。
在新笔记本的第一个单元格中,添加以下代码,然后运行该单元格,该单元格调用%皮普魔法。这个魔术安装pytest.
%皮普
安装pytest
在第二个单元格中,添加以下代码,替换< my-repo-name >,然后运行单元格。结果显示通过和失败的单元测试。
< my-repo-name >
进口pytest进口操作系统进口sysrepo_name=“< my-repo-name >”#获取该笔记本的路径,例如“/Workspace/Repos/{username}/{repo-name}”。notebook_path=dbutils.笔记本.entry_point.getDbutils().笔记本().getContext().notebookPath().得到()#获取repo的根目录名。repo_root=操作系统.路径.目录名(操作系统.路径.目录名(notebook_path))#准备从repo运行pytest。操作系统.是指(f“/工作区/{repo_root}/{repo_name}")打印(操作系统.getcwd())#跳过在只读文件系统上写pyc文件。sys.dont_write_bytecode=真正的#运行pytest。retcode=pytest.主要([“。”,“v”,“p”,“没有:cacheprovider”])如果有任何测试失败,则单元格执行失败。断言retcode= =0," pytest调用失败。详见日志。”
在上面的文件夹中创建一个R笔记本test_myfunctions.r文件在您的回购,并添加以下内容。
在第一个单元格中,添加以下代码,然后运行单元格,该单元格调用install.packages函数。这个函数安装testthat.
install.packages
install.packages(“testthat”)
在第二个单元格中,添加以下代码,然后运行该单元格。结果显示通过和失败的单元测试。
图书馆(testthat)源(“myfunctions.r”)test_dir(“。”,记者=“龙头”)
运行上一节中笔记本中的第一个和第二个单元格。结果显示通过和失败的单元测试。
运行上一节中笔记本中的三个单元格。结果显示每个单元测试是通过还是失败。
如果运行单元测试后不再需要该视图,可以删除该视图。要删除此视图,可以将以下代码添加到前面一个笔记本中的一个新单元格中,然后只运行该单元格。
下降视图view_diamonds;
提示
您可以在集群中查看notebook运行的结果(包括单元测试结果)司机日志.您还可以为集群指定一个位置日志交付.
您可以设置持续集成和持续交付或部署(CI/CD)系统,例如GitHub Actions,以便在代码更改时自动运行单元测试。举个例子,请参见笔记本电脑软件工程最佳实践.
pytest主页
Pytest操作指南
Pytest参考指南
笔记本电脑软件工程最佳实践
使用Databricks的IDE
testthat主页
测试该函数引用
——scalate主页
ScalaTest用户指南
ScalaTest的Scaladoc文档
创建函数
创建视图