单元测试的笔记本

进口pyspark从pyspark.sql进口SparkSession从pyspark.sql.functions进口上校#因为这个文件不是一个砖笔记本,你#必须创建一个火花会话。砖的笔记本#默认为您创建一个火花会话。火花=SparkSession。构建器\。浏览器名称(“诚信测试”)\。getOrCreate()#指定数据库中指定的表存在吗?deftableExists(的表,dbName):返回火花。目录。tableExists(f”{dbName}。{的表}”)#在给定DataFrame指定的列存在吗?defcolumnExists(dataFrame,columnName):如果columnName在dataFrame。列:返回真正的其他的:返回假#有多少行中的指定值指定列在给定DataFrame #吗?defnumRowsInColumnForValue(dataFrame,columnName,columnValue):df=dataFrame。过滤器(上校(columnName)= =columnValue)返回df。数()

从myfunction进口*的表=“钻石”dbName=“默认”columnName=“清晰”columnValue=“VVS2”#如果表存在于指定的数据库……如果tableExists(的表,dbName):df=火花。sql(f“SELECT *{dbName}。{的表}”)#和指定的列存在于表…如果columnExists(df,columnName):#然后报告中的指定值的行数,列。numRows=numRowsInColumnForValue(df,columnName,columnValue)打印(f“有{numRows}行”{的表}“在那里”{columnName}“=”{columnValue}’。”)其他的:打印(f“列{columnName}“不存在表”{的表}”模式(数据库)”{dbName}’。”)其他的:打印(f“表{的表}不存在的模式(数据库)”{dbName}’。”)

进口pytest进口pyspark从myfunction进口*从pyspark.sql进口SparkSession从pyspark.sql.types进口StructType,StructField,IntegerType,FloatType,StringType的表=“钻石”dbName=“默认”columnName=“清晰”columnValue=“SI2”#因为这个文件不是一个砖笔记本,你#必须创建一个火花会话。砖的笔记本#默认为您创建一个火花会话。火花=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

进口pytest进口操作系统进口sysrepo_name=“< my-repo-name >”#这个笔记本的路径,例如“/工作区/回购/ {username} / {repo-name}”。notebook_path=dbutils。笔记本。entry_point。getDbutils()。笔记本()。getContext()。notebookPath()。得到()#回购的根目录的名字。repo_root=操作系统。路径。目录名(操作系统。路径。目录名(notebook_path))#准备运行pytest回购。操作系统。是指(f“/工作区/{repo_root}/{repo_name}”)打印(操作系统。getcwd())#跳过写佩克一个只读的文件系统上的文件。sys。dont_write_bytecode=真正的# pytest运行。retcode=pytest。主要([“。”,“v”,“p”,“没有:cacheprovider”])#失败细胞执行是否有测试失败。断言retcode= =0,“pytest调用失败了。有关详细信息,请参阅日志”。