开始
加载和管理数据
处理数据
政府
引用和资源
2023年2月22日更新
给我们反馈
请注意
本文涵盖了詹金斯,既不提供也不支持数据砖。联系供应商,看詹金斯帮助。
持续集成和持续交付(CI / CD)是指开发和交付软件的过程简而言之,通过使用自动化管道频繁的周期。这决不是一个新进程,无处不在的几十年来在传统的软件工程,它正成为一个越来越必要的过程工程和数据科学团队。为了使数据产品是有价值的,他们必须及时交付。此外,消费者必须有信心在这些产品结果的有效性。通过自动化的构建、测试和部署代码,开发团队能够提供更频繁地发布和可靠地比手工流程仍普遍在许多工程和数据科学团队。
持续集成始于实践您提交您的代码的一些频率源代码存储库中的一个分支。然后每个提交合并与其他开发人员的承诺,以确保没有冲突。变化是进一步验证通过创建一个构建和运行自动化测试与构建。这个过程最终导致工件,或部署包,最终将部署到目标环境中,在这种情况下一个砖的工作区。
虽然可以根据您的需要,不同的典型配置数据砖管道包括以下步骤:
持续集成:
代码
开发代码和单元测试砖笔记本或使用一个外部的IDE。
手动运行测试。
提交代码和测试一个git分支。
构建
收集新的和更新的代码和测试。
运行自动化测试。
构建Apache火花代码库和儿童。
发布:生成一个释放工件。
持续交付:
部署
部署的笔记本。
配置库。
测试:运行自动化测试并报告结果。
操作:通过编程工程进度数据,分析和机器学习的工作流。
第一个步骤在设计CI / CD决定代码提交和管道的分支策略来管理新的和更新的代码的开发和集成代码目前在生产中没有不良影响。这个决定需要选择一个版本控制系统包含代码和促进的促销代码。砖支持集成的GitHub, Bitbucket都,允许你提交笔记本git存储库。
如果你的版本控制系统不是那些支持通过直接笔记本集成,或者如果你想要更大的灵活性和控制比自助git集成,您可以使用砖CLI出口笔记本和提交本地机器。这个脚本应该运行在一个本地git存储库设置为与适当的远程存储库同步。当执行时,这个脚本应该:
检查所需的分支。
把新变化从远程分支。
出口笔记本电脑使用砖CLI从砖工作区。
提示用户提交消息或使用默认的如果一个人没有提供。
提交更新后的笔记本到本地分支。
推动远程分支的变化。
下面的脚本执行以下步骤:
git checkout <分支> git拉砖工作区export_dir——概要文件<档案> - o <路径>。/工作区dt=”日期' + Y % - % - % d % H: % m: % S的”msg_default=“DB出口dt美元”读- p“进入提交评论msg_default美元):“味精味精=$ {味精:-msg_default美元}回声美元味精git添加。git commit - m“<提交消息>”git推
如果你喜欢在砖开发IDE而不是笔记本电脑,您可以使用内置的VCS集成特性现代IDE或git CLI提交代码。
砖提供砖连接,一个SDK,连接ide砖集群。这是开发库时特别有用,因为它允许您砖集群上运行代码和单元测试,而无需部署代码。看到砖连接限制确定是否支持你的用例。
砖现在建议你使用dbx由砖实验室为当地的发展而不是砖连接。
根据您的分支策略和推广过程中,点CI / CD管道将启动建立会有所不同。然而,承诺从各种贡献者的代码最终将合并成一个指定的分支是构建和部署。分支管理步骤运行之外的砖,使用提供的版本控制系统的接口。
有许多CI / CD工具可以用来管理和执行你的管道。本文说明了如何使用詹金斯自动化服务器。CI / CD是一个设计模式,所以在这篇文章中概述的步骤和阶段应该转移一些更改管道在每个工具定义语言。此外,大部分的代码在这个例子中管道运行标准的Python代码,您可以调用其他工具。
詹金斯使用一个主服务协调和一个到多个执行代理。在本例中您使用默认永久代理节点包含在詹金斯服务器。您必须手动安装以下工具和软件包所需的代理上的管道,在这种情况下,詹金斯服务器:
Conda:一个开放bob下载地址源码的Python环境管理体系。
Python 3.7:用于运行测试,构建一个部署轮,并执行部署脚本。Python的版本是很重要的,因为测试要求的Python版本运行在集群代理应该匹配的砖。这个示例使用砖7.3运行时,其中包括Python 3.7。
Python库:请求,databricks-connect,databricks-cli,pytest。
请求
databricks-connect
databricks-cli
pytest
詹金斯提供了一些不同的项目类型创建CI / CD管道。这个例子实现了一个詹金斯管道。詹金斯管道提供一个接口来定义阶段管道使用Groovy詹金斯代码来调用和配置插件。
你写一个管道定义在一个文本文件(称为Jenkinsfile)进而检查项目的版本控制存储库。有关更多信息,请参见詹金斯管道。下面是一个示例管道:
/ / Jenkinsfile节点{defGITREPO=" / var / lib / jenkins /工作区/ $ {env.JOB_NAME} "defGITREPOREMOTE=“https://github.com/ <回购>”defGITHUBCREDID=“< github-token >”defCURRENTRELEASE=“<版本>”defDBTOKEN=“< databricks-token >”defDBURL=“https:// < databricks-instance >”defSCRIPTPATH=“$ {GITREPO} /自动化/部署”defNOTEBOOKPATH=“$ {GITREPO} /工作空间”defLIBRARYPATH=“$ {GITREPO} /库”defBUILDPATH=" $ {GITREPO} /构建/ {env.JOB_NAME} - {env.BUILD_NUMBER}”美元defOUTFILEPATH=“$ {BUILDPATH} /验证/输出”defTESTRESULTPATH=" $ {BUILDPATH} /验证/报告/ junit”defWORKSPACEPATH=“/共享/ <路径>”defDBFSPATH=“dbfs: < dbfs-path >”defCLUSTERID=“< cluster-id >”defCONDAPATH=“< conda-path >”defCONDAENV=“< conda-env >”阶段(“设置”){withCredentials([字符串(credentialsId:DBTOKEN,变量:“令牌”))){上海" " # ! / bin / bash#配置Conda部署和测试环境源$ {CONDAPATH} / bin / $ {CONDAENV}激活#为部署配置砖CLIecho " $ {DBURL}美元令牌”|砖配置——令牌#配置数据砖连接进行测试echo " $ {DBURL}美元的令牌$ {CLUSTERID}015001年“| databricks-connect配置”“”}}阶段(“签出”){/ /用于显示目的回声“把$ {CURRENTRELEASE}分支从Github”git分支机构:CURRENTRELEASE,credentialsId:GITHUBCREDID,url:GITREPOREMOTE}阶段(“运行单元测试”){试一试{上海" " # ! / bin / bash#让Conda环境测试源$ {CONDAPATH} / bin / $ {CONDAENV}激活# Python测试库python3 - m pytest——junit-xml = $ {TESTRESULTPATH} / TEST-libout。xml $ {LIBRARYPATH} * / python / dbxdemo /测试。py | |真”“”}抓(犯错){一步([美元的类:“JUnitResultArchiver”,检测结果:”——junit-xml = $ {TESTRESULTPATH} /测试- * . xml”])如果(currentBuild。结果= =“不稳定”)currentBuild。结果=“失败”扔犯错}}阶段(“包”){上海" " # ! / bin / bash#让Conda环境测试源$ {CONDAPATH} / bin / $ {CONDAENV}激活#包Python库cd $ {LIBRARYPATH} / python / dbxdemopython3设置。py sdist bdist_wheel”“”}阶段(建立工件的){上海”““mkdir - p $ {BUILDPATH} /工作区mkdir - p $ {BUILDPATH} /图书馆/ pythonmkdir - p $ {BUILDPATH} /验证/输出#被修改的文件git diff——名义diff-filter AMR交头接耳^ 1 = | xargs -“{}”cp -家长- r“{}”$ {BUILDPATH}#获取打包填词找到$ {LIBRARYPATH} - name”*。whl' | xargs -I '{}' cp '{}' ${BUILDPATH}/Libraries/python/#生成工件焦油-czvf构建/ latest_build.tar。广州$ {BUILDPATH}”“”archiveArtifacts构件:“构建/ latest_build.tar.gz”}阶段(“部署”){上海" " # ! / bin / bash#让Conda环境测试源$ {CONDAPATH} / bin / $ {CONDAENV}激活#使用砖CLI部署的笔记本砖工作区import_dir $ {BUILDPATH} /工作区$ {WORKSPACEPATH}dbfs cp - r $ {BUILDPATH} /图书馆/ python $ {DBFSPATH}”“”withCredentials([字符串(credentialsId:DBTOKEN,变量:“令牌”))){上海" " # ! / bin / bash#得到图书馆的空间分隔的列表填词= \ $(找到$ {BUILDPATH} /图书馆/ python / * - name”。whl年代# | sed。* / # #”|粘贴-标准差" ")#脚本卸载,重启如果需要& instsall图书馆python3 $ {SCRIPTPATH} / installWhlLibrary。py——工作区= $ {DBURL} \——令牌= $令牌\——clusterid = $ {clusterid} \——libs = \ libs美元\——dbfspath = $ {dbfspath}”“”}}阶段(“运行集成测试”){withCredentials([字符串(credentialsId:DBTOKEN,变量:“令牌”))){上海”““python3 $ {SCRIPTPATH} / executenotebook。py——工作区= $ {DBURL} \——令牌= $令牌\——clusterid = $ {clusterid} \——localpath = $ {NOTEBOOKPATH} \ /验证——workspacepath = $ {workspacepath} \ /验证——outfilepath = $ {outfilepath}”“”}上海" " " sed - i - e的# ENV # $ {OUTFILEPATH} g’$ {SCRIPTPATH} / evaluatenotebookruns.pypython3 - m pytest——junit-xml = $ {TESTRESULTPATH} / TEST-notebookout。xml $ {SCRIPTPATH} / evaluatenotebookruns。py | |真”“”}阶段(“报告测试结果”){上海”““找到$ {OUTFILEPATH} - name”*。json' -exec gzip --verbose {} \\;触摸$ {TESTRESULTPATH} /测试- * . xml”“”junit“* * /报告/ junit / * . xml”}}
本文的其余部分讨论了管道的每一步。
您可以定义环境变量允许管道阶段中使用不同的管道。
作为一个安全最佳实践,验证自动化工具时,系统中,脚本和应用程序,砖属于建议您使用访问令牌服务主体而不是用户工作区。为服务主体,创建访问令牌管理服务主体的访问令牌。
GITREPO:本地路径git存储库根
GITREPO
GITREPOREMOTE:git存储库Web URL
GITREPOREMOTE
GITHUBCREDID:詹金斯凭据ID GitHub个人访问令牌
GITHUBCREDID
CURRENTRELEASE:部署分支
CURRENTRELEASE
DBTOKEN:詹金斯凭据ID砖个人访问令牌
DBTOKEN
DBURL:Web URL砖工作区
DBURL
SCRIPTPATH:本地路径自动化脚本的git项目目录
SCRIPTPATH
NOTEBOOKPATHgit项目目录:本地路径笔记本
NOTEBOOKPATH
LIBRARYPATHgit项目:本地路径目录库代码或其他DBFS代码
LIBRARYPATH
BUILDPATH:本地路径目录构建工件
BUILDPATH
OUTFILEPATH:本地路径从自动测试生成的JSON结果文件
OUTFILEPATH
TESTRESULTPATH:本地路径目录Junit测试结果总结
TESTRESULTPATH
WORKSPACEPATH笔记本电脑:砖工作区路径
WORKSPACEPATH
DBFSPATH:砖DBFS路径库和儿童的代码
DBFSPATH
CLUSTERID:砖集群ID来运行测试
CLUSTERID
CONDAPATH:Conda安装路径
CONDAPATH
CONDAENV:名字Conda环境包含构建依赖库的
CONDAENV
在设置阶段您配置数据砖CLI和砖与连接信息。
设置
defGITREPO=" / var / lib / jenkins /工作区/ $ {env.JOB_NAME} "defGITREPOREMOTE=“https://github.com/ <回购>”defGITHUBCREDID=“< github-token >”defCURRENTRELEASE=“<版本>”defDBTOKEN=“< databricks-token >”defDBURL=“https:// < databricks-instance >”defSCRIPTPATH=“$ {GITREPO} /自动化/部署”defNOTEBOOKPATH=“$ {GITREPO} /工作空间”defLIBRARYPATH=“$ {GITREPO} /库”defBUILDPATH=" $ {GITREPO} /构建/ {env.JOB_NAME} - {env.BUILD_NUMBER}”美元defOUTFILEPATH=“$ {BUILDPATH} /验证/输出”defTESTRESULTPATH=" $ {BUILDPATH} /验证/报告/ junit”defWORKSPACEPATH=“/共享/ <路径>”defDBFSPATH=“dbfs: < dbfs-path >”defCLUSTERID=“< cluster-id >”defCONDAPATH=“< conda-path >”defCONDAENV=“< conda-env >”阶段(“设置”){withCredentials([字符串(credentialsId:DBTOKEN,变量:“令牌”))){上海" " # ! / bin / bash#配置Conda部署和测试环境源$ {CONDAPATH} / bin / $ {CONDAENV}激活#为部署配置砖CLIecho " $ {DBURL}美元令牌”|砖配置——令牌#配置数据砖连接进行测试echo " $ {DBURL}美元的令牌$ {CLUSTERID}015001年“| databricks-connect配置”“”}}
的结帐阶段从指定下载代码分支代理执行代理使用一个詹金斯插件:
结帐
阶段(“签出”){/ /用于显示目的回声“把$ {CURRENTRELEASE}分支从Github”git分支机构:CURRENTRELEASE,credentialsId:GITHUBCREDID,url:GITREPOREMOTE}
有几个不同的选项在决定如何单元测试代码。为图书馆开发的代码外砖笔记本,这个过程就像传统的软件开发实践。你编写一个单元测试使用的测试框架,像Pythonpytest模块,和JUnit-formatted XML文件存储测试结果。
砖过程不同,被测试的代码是Apache引发火花集群上执行代码的目的通常运行在本地或在这种情况下砖。为了满足这种需求,你使用砖连接。自从SDK早些时候配置,不需要修改测试代码砖集群上执行测试。你安装了砖连接Conda虚拟环境中。一旦激活Conda环境,测试执行使用Python工具,pytest,你提供测试和由此产生的输出文件的位置。
阶段(“运行单元测试”){试一试{上海" " # ! / bin / bash#让Conda环境测试源$ {CONDAPATH} / bin / $ {CONDAENV}激活# Python测试库python3 - m pytest——junit-xml = $ {TESTRESULTPATH} / TEST-libout。xml $ {LIBRARYPATH} * / python / dbxdemo /测试。py | |真”“”}抓(犯错){一步([美元的类:“JUnitResultArchiver”,检测结果:”——junit-xml = $ {TESTRESULTPATH} /测试- * . xml”])如果(currentBuild。结果= =“不稳定”)currentBuild。结果=“失败”扔犯错}}
下面的代码片段是一个库函数,可能是安装在一个砖集群。这是一个简单的函数,添加一个新列,由文字组成,一个Apache DataFrame火花。
# addcol.py进口pyspark.sql.functions作为Fdefwith_status(df):返回df。withColumn(“状态”,F。点燃(“检查”))
这个测试通过模拟DataFrame对象with_status中定义的函数,addcol.py。结果然后比较DataFrame对象包含预期的值。如果值匹配,在这种情况下,测试通过。
with_status
addcol.py
# test-addcol.py进口pytest从dbxdemo.spark进口get_spark从dbxdemo.appendcol进口with_status类TestAppendCol(对象):deftest_with_status(自我):source_data=((“波拉”,“白色”,“paula.white@example.com”),(“约翰。”,“贝尔”,“john.baer@example.com”)]source_df=get_spark()。createDataFrame(source_data,(“first_name”,“last_name”,“电子邮件”])actual_df=with_status(source_df)expected_data=((“波拉”,“白色”,“paula.white@example.com”,“检查”),(“约翰。”,“贝尔”,“john.baer@example.com”,“检查”)]expected_df=get_spark()。createDataFrame(expected_data,(“first_name”,“last_name”,“电子邮件”,“状态”])断言(expected_df。收集()= =actual_df。收集())
在包阶段,你包库代码到一个Python轮。
包
阶段(“包”){上海" " # ! / bin / bash#让Conda环境测试源$ {CONDAPATH} / bin / $ {CONDAENV}激活#包Python库cd $ {LIBRARYPATH} / python / dbxdemopython3设置。py sdist bdist_wheel”“”}
建立一个部署构件砖包括收集所有新的或更新的代码部署到适当的砖的环境。在构建工件阶段添加笔记本代码部署到工作区,whl构建过程生成的库,以及测试结果总结,用于存档。要做到这一点,你使用gitdiff标记所有新文件已经包含在最新git合并。这只是一个例子的方法,实现在你的管道可能有所不同,但目标是将所有文件用于当前版本。
构建工件
whl
gitdiff
阶段(建立工件的){上海”““mkdir - p $ {BUILDPATH} /工作区mkdir - p $ {BUILDPATH} /图书馆/ pythonmkdir - p $ {BUILDPATH} /验证/输出#被修改的文件git diff——名义diff-filter AMR交头接耳^ 1 = | xargs -“{}”cp -家长- r“{}”$ {BUILDPATH}#获取打包填词找到$ {LIBRARYPATH} - name”*。whl' | xargs -I '{}' cp '{}' ${BUILDPATH}/Libraries/python/#生成工件焦油-czvf构建/ latest_build.tar。广州$ {BUILDPATH}”“”archiveArtifacts构件:“构建/ latest_build.tar.gz”}
在部署阶段使用砖CLI,就像早些时候使用的砖连接模块,安装在你Conda环境,所以你必须为这个shell会话激活它。你使用工作区CLI和DBFS CLI上传笔记本和库,分别为:
砖工作区import_dir <当地的构建路径> <远程工作区路径> dbfs cp - r <当地的构建路径> <远程dbfs路径>
阶段(“部署”){上海" " # ! / bin / bash#让Conda环境测试源$ {CONDAPATH} / bin / $ {CONDAENV}激活#使用砖CLI部署的笔记本砖工作区import_dir $ {BUILDPATH} /工作区$ {WORKSPACEPATH}dbfs cp - r $ {BUILDPATH} /图书馆/ python $ {DBFSPATH}”“”withCredentials([字符串(credentialsId:DBTOKEN,变量:“令牌”))){上海" " # ! / bin / bash#得到图书馆的空间分隔的列表填词= \ $(找到$ {BUILDPATH} /图书馆/ python / * - name”。whl年代# | sed。* / # #”|粘贴-标准差" ")#脚本卸载,重启如果需要& instsall图书馆python3 $ {SCRIPTPATH} / installWhlLibrary。py——工作区= $ {DBURL} \——令牌= $令牌\——clusterid = $ {clusterid} \——libs = \ libs美元\——dbfspath = $ {dbfspath}”“”}}
安装新版本库的砖集群要求您首先卸载现有的图书馆。要做到这一点,你调用砖REST API在一个Python脚本执行以下步骤:
检查是否安装了图书馆。
卸载图书馆。
重新启动集群是否进行卸载。
等到再次集群运行之前。
安装该库。
# installWhlLibrary.py# ! / usr / bin / python3进口json进口请求进口sys进口getopt进口时间def主要():工作空间=”令牌=”clusterid=”填词=”dbfspath=”试一试:选择,arg游戏=getopt。getopt(sys。argv(1:,“hstcld”,(“工作区= ',“令牌= ',“clusterid = ',“填词=”,“dbfspath = '])除了getopt。GetoptError:打印(“installWhlLibrary。py -s -t -c -l -d ')sys。退出(2)为选择,参数在选择:如果选择= =“- h”:打印(“installWhlLibrary。py -s -t -c -l -d ')sys。退出()elif选择在(“s”,“——工作区”):工作空间=参数elif选择在(“t”,“——令牌”):令牌=参数elif选择在(“c”,“——clusterid”):clusterid=参数elif选择在(“- l”,“那——libs”):填词=参数elif选择在(“- d”,“——dbfspath”):dbfspath=参数打印(“s”+工作空间)打印(“- t是”+令牌)打印(“c”+clusterid)打印(“- l是”+填词)打印(“- d”+dbfspath)libslist=填词。分裂()#卸载图书馆如果存在于集群我=0为自由在libslist:dbfslib=dbfspath+自由打印(dbfslib+“:”+getLibStatus(工作空间,令牌,clusterid,dbfslib))如果(getLibStatus(工作空间,令牌,clusterid,dbfslib)! =“没有找到”):打印(dbfslib+”存在。卸载。”)我=我+1值={“cluster_id”:clusterid,“库”:[{“whl”:dbfslib}]}分别地=请求。帖子(工作空间+/ api / 2.0 /图书馆/卸载的,数据=json。转储(值),身份验证=(“令牌”,令牌))runjson=分别地。文本d=json。加载(runjson)打印(dbfslib+”后,“+getLibStatus(工作空间,令牌,clusterid,dbfslib))#如果库卸载重启如果我>0:值={“cluster_id”:clusterid}打印(重新启动集群:“+clusterid)分别地=请求。帖子(工作空间+“/ api / 2.0 /集群/重启”,数据=json。转储(值),身份验证=(“令牌”,令牌))restartjson=分别地。文本打印(restartjson)p=0等待=真正的而等待:时间。睡眠(30.)clusterresp=请求。得到(工作空间+' / api / 2.0 /集群/ ? cluster_id = '+clusterid,身份验证=(“令牌”,令牌))clusterjson=clusterresp。文本jsonout=json。加载(clusterjson)current_state=jsonout(“状态”]打印(clusterid+”状态:“+current_state)如果current_state在(“奔跑”,“INTERNAL_ERROR”,“跳过”]或p> =10:打破p=p+1#安装库为自由在libslist:dbfslib=dbfspath+自由打印(“安装”+dbfslib)值={“cluster_id”:clusterid,“库”:[{“whl”:dbfslib}]}分别地=请求。帖子(工作空间+/ api / 2.0 /图书馆/安装的,数据=json。转储(值),身份验证=(“令牌”,令牌))runjson=分别地。文本d=json。加载(runjson)打印(dbfslib+”后,“+getLibStatus(工作空间,令牌,clusterid,dbfslib))defgetLibStatus(工作空间,令牌,clusterid,dbfslib):分别地=请求。得到(工作空间+' / api / 2.0 /图书馆/集群状态? cluster_id = '+clusterid,身份验证=(“令牌”,令牌))libjson=分别地。文本d=json。加载(libjson)如果(d。得到(“library_statuses”)):状态=d(“library_statuses”]为状态在状态:如果(状态(“图书馆”]。得到(“whl”)):如果(状态(“图书馆”][“whl”]= =dbfslib):返回状态(“状态”]其他的:返回“没有找到”其他的:#没有发现库返回“没有找到”如果__name__= =“__main__ ':主要()
一旦工件被部署,重要的是要运行集成测试,以确保所有的代码一起工作在新的环境。要做到这一点,你可以运行一个笔记本包含断言测试部署。在这种情况下,您正在使用相同的测试中使用单元测试,但现在是进口安装appendcol图书馆的whl这只是在集群上安装。
appendcol
自动化测试,包括你的CI / CD管道,使用砖REST API的笔记本从詹金斯服务器运行。这允许您检查笔记本运行是否通过或失败pytest。如果笔记本的断言失败,这将是由REST API返回的JSON输出所示,随后在JUnit测试结果。
阶段(“运行集成测试”){withCredentials([字符串(credentialsId:DBTOKEN,变量:“令牌”))){上海”““python3 $ {SCRIPTPATH} / executenotebook。py——工作区= $ {DBURL} \——令牌= $令牌\——clusterid = $ {clusterid} \——localpath = $ {NOTEBOOKPATH} \ /验证——workspacepath = $ {workspacepath} \ /验证——outfilepath = $ {outfilepath}”“”}上海" " " sed - i - e的# ENV # $ {OUTFILEPATH} g’$ {SCRIPTPATH} / evaluatenotebookruns.pypython3 - m pytest——junit-xml = $ {TESTRESULTPATH} / TEST-notebookout。xml $ {SCRIPTPATH} / evaluatenotebookruns。py | |真”“”}
这个阶段调用两个Python自动化脚本。第一个脚本,executenotebook.py,笔记本使用创建并触发一次运行(帖子/ /运行/提交工作)端点提交一个匿名的工作。因为这个端点是异步的,它使用REST调用返回的任务ID最初调查工作的状态。一旦工作完成,JSON输出保存到指定的路径在调用函数参数传递。
executenotebook.py
帖子/ /运行/提交工作
# executenotebook.py# ! / usr / bin / python3进口json进口请求进口操作系统进口sys进口getopt进口时间def主要():工作空间=”令牌=”clusterid=”localpath=”workspacepath=”outfilepath=”试一试:选择,arg游戏=getopt。getopt(sys。argv(1:,“海关:t: c:靠”,(“工作区= ',“令牌= ',“clusterid = ',“localpath = ',“workspacepath = ',“outfilepath = '])除了getopt。GetoptError:打印(“executenotebook。py -s -t -c -l -w -o )')sys。退出(2)为选择,参数在选择:如果选择= =“- h”:打印(“executenotebook。py -s -t -c -l -w -o ')sys。退出()elif选择在(“s”,“——工作区”):工作空间=参数elif选择在(“t”,“——令牌”):令牌=参数elif选择在(“c”,“——clusterid”):clusterid=参数elif选择在(“- l”,“——localpath”):localpath=参数elif选择在(“- w”,“——workspacepath”):workspacepath=参数elif选择在(“o”,“——outfilepath”):outfilepath=参数打印(“s”+工作空间)打印(“- t是”+令牌)打印(“c”+clusterid)打印(“- l是”+localpath)打印(“- w是”+workspacepath)打印(“- o '+outfilepath)#从走本地路径生成数组笔记本电脑=[]为路径,子目录,文件在操作系统。走(localpath):为的名字在文件:fullpath=路径+' / '+的名字#删除localpath回购,但保持工作区路径fullworkspacepath=workspacepath+路径。取代(localpath,”)的名字,file_extension=操作系统。路径。splitext(fullpath)如果file_extension。较低的()在(. scala的,. sql的,“r”,. py的]:行=(fullpath,fullworkspacepath,1]笔记本电脑。附加(行)#运行每个元素列表为笔记本在笔记本电脑:nameonly=操作系统。路径。basename(笔记本(0])workspacepath=笔记本(1]的名字,file_extension=操作系统。路径。splitext(nameonly)# workpath删除扩展fullworkspacepath=workspacepath+' / '+的名字打印(的工作:+fullworkspacepath)值={“run_name”:的名字,“existing_cluster_id”:clusterid,“timeout_seconds”:3600年,“notebook_task”:{“notebook_path”:fullworkspacepath}}分别地=请求。帖子(工作空间+' / api / 2.0 /工作/运行/提交的,数据=json。转储(值),身份验证=(“令牌”,令牌))runjson=分别地。文本打印(”runjson:“+runjson)d=json。加载(runjson)runid=d(“run_id”]我=0等待=真正的而等待:时间。睡眠(10)jobresp=请求。得到(工作空间+' / api / 2.0 /工作/运行/ ? run_id = '+str(runid),数据=json。转储(值),身份验证=(“令牌”,令牌))jobjson=jobresp。文本打印(”jobjson:“+jobjson)j=json。加载(jobjson)current_state=j(“状态”][“life_cycle_state”]runid=j(“run_id”]如果current_state在(“终止”,“INTERNAL_ERROR”,“跳过”]或我> =12:打破我=我+1如果outfilepath! =”:文件=开放(outfilepath+' / '+str(runid)+. json的,' w ')文件。写(json。转储(j))文件。关闭()如果__name__= =“__main__ ':主要()
第二个脚本,evaluatenotebookruns.py,定义了test_job_run函数,解析和评价JSON来确定在笔记本通过还是失败的断言语句。一个额外的测试,test_performance,捕获测试运行比预期更长的时间。
evaluatenotebookruns.py
test_job_run
test_performance
# evaluatenotebookruns.py进口unittest进口json进口一团进口操作系统类TestJobOutput(unittest。TestCase):test_output_path=“# ENV #”deftest_performance(自我):路径=自我。test_output_path状态=[]为文件名在一团。一团(操作系统。路径。加入(路径,‘* . json‘)):打印(的评价:“+文件名)数据=json。负载(开放(文件名))持续时间=数据(“execution_duration”]如果持续时间>100000年:状态=“失败”其他的:状态=“成功”状态。附加(状态)自我。assertFalse(“失败”在状态)deftest_job_run(自我):路径=自我。test_output_path状态=[]为文件名在一团。一团(操作系统。路径。加入(路径,‘* . json‘)):打印(的评价:“+文件名)数据=json。负载(开放(文件名))状态=数据(“状态”][“result_state”]状态。附加(状态)自我。assertFalse(“失败”在状态)如果__name__= =“__main__ ':unittest。主要()
你见过在单元测试阶段,使用pytest运行测试并生成结果摘要。
JSON结果存档和詹金斯使用发布测试结果junit詹金斯插件。这使您能够可视化报告和仪表板与构建过程的状态有关。
junit
阶段(“报告测试结果”){上海”““找到$ {OUTFILEPATH} - name”*。json' -exec gzip --verbose {} \\;触摸$ {TESTRESULTPATH} /测试- * . xml”“”junit“* * /报告/ junit / * . xml”}
此时,CI / CD管道已完成集成和部署周期。通过自动化这个过程,你可以确保你的代码测试和部署了一个高效的、一致的、可重复的过程。