diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..261eeb9 --- /dev/null +++ b/LICENSE @@ -0,0 +1,201 @@ + Apache License + Version 2.0, January 2004 + http://www.apache.org/licenses/ + + TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION + + 1. Definitions. + + "License" shall mean the terms and conditions for use, reproduction, + and distribution as defined by Sections 1 through 9 of this document. + + "Licensor" shall mean the copyright owner or entity authorized by + the copyright owner that is granting the License. + + "Legal Entity" shall mean the union of the acting entity and all + other entities that control, are controlled by, or are under common + control with that entity. For the purposes of this definition, + "control" means (i) the power, direct or indirect, to cause the + direction or management of such entity, whether by contract or + otherwise, or (ii) ownership of fifty percent (50%) or more of the + outstanding shares, or (iii) beneficial ownership of such entity. + + "You" (or "Your") shall mean an individual or Legal Entity + exercising permissions granted by this License. + + "Source" form shall mean the preferred form for making modifications, + including but not limited to software source code, documentation + source, and configuration files. + + "Object" form shall mean any form resulting from mechanical + transformation or translation of a Source form, including but + not limited to compiled object code, generated documentation, + and conversions to other media types. + + "Work" shall mean the work of authorship, whether in Source or + Object form, made available under the License, as indicated by a + copyright notice that is included in or attached to the work + (an example is provided in the Appendix below). + + "Derivative Works" shall mean any work, whether in Source or Object + form, that is based on (or derived from) the Work and for which the + editorial revisions, annotations, elaborations, or other modifications + represent, as a whole, an original work of authorship. For the purposes + of this License, Derivative Works shall not include works that remain + separable from, or merely link (or bind by name) to the interfaces of, + the Work and Derivative Works thereof. + + "Contribution" shall mean any work of authorship, including + the original version of the Work and any modifications or additions + to that Work or Derivative Works thereof, that is intentionally + submitted to Licensor for inclusion in the Work by the copyright owner + or by an individual or Legal Entity authorized to submit on behalf of + the copyright owner. For the purposes of this definition, "submitted" + means any form of electronic, verbal, or written communication sent + to the Licensor or its representatives, including but not limited to + communication on electronic mailing lists, source code control systems, + and issue tracking systems that are managed by, or on behalf of, the + Licensor for the purpose of discussing and improving the Work, but + excluding communication that is conspicuously marked or otherwise + designated in writing by the copyright owner as "Not a Contribution." + + "Contributor" shall mean Licensor and any individual or Legal Entity + on behalf of whom a Contribution has been received by Licensor and + subsequently incorporated within the Work. + + 2. Grant of Copyright License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + copyright license to reproduce, prepare Derivative Works of, + publicly display, publicly perform, sublicense, and distribute the + Work and such Derivative Works in Source or Object form. + + 3. Grant of Patent License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + (except as stated in this section) patent license to make, have made, + use, offer to sell, sell, import, and otherwise transfer the Work, + where such license applies only to those patent claims licensable + by such Contributor that are necessarily infringed by their + Contribution(s) alone or by combination of their Contribution(s) + with the Work to which such Contribution(s) was submitted. If You + institute patent litigation against any entity (including a + cross-claim or counterclaim in a lawsuit) alleging that the Work + or a Contribution incorporated within the Work constitutes direct + or contributory patent infringement, then any patent licenses + granted to You under this License for that Work shall terminate + as of the date such litigation is filed. + + 4. Redistribution. You may reproduce and distribute copies of the + Work or Derivative Works thereof in any medium, with or without + modifications, and in Source or Object form, provided that You + meet the following conditions: + + (a) You must give any other recipients of the Work or + Derivative Works a copy of this License; and + + (b) You must cause any modified files to carry prominent notices + stating that You changed the files; and + + (c) You must retain, in the Source form of any Derivative Works + that You distribute, all copyright, patent, trademark, and + attribution notices from the Source form of the Work, + excluding those notices that do not pertain to any part of + the Derivative Works; and + + (d) If the Work includes a "NOTICE" text file as part of its + distribution, then any Derivative Works that You distribute must + include a readable copy of the attribution notices contained + within such NOTICE file, excluding those notices that do not + pertain to any part of the Derivative Works, in at least one + of the following places: within a NOTICE text file distributed + as part of the Derivative Works; within the Source form or + documentation, if provided along with the Derivative Works; or, + within a display generated by the Derivative Works, if and + wherever such third-party notices normally appear. The contents + of the NOTICE file are for informational purposes only and + do not modify the License. You may add Your own attribution + notices within Derivative Works that You distribute, alongside + or as an addendum to the NOTICE text from the Work, provided + that such additional attribution notices cannot be construed + as modifying the License. + + You may add Your own copyright statement to Your modifications and + may provide additional or different license terms and conditions + for use, reproduction, or distribution of Your modifications, or + for any such Derivative Works as a whole, provided Your use, + reproduction, and distribution of the Work otherwise complies with + the conditions stated in this License. + + 5. Submission of Contributions. Unless You explicitly state otherwise, + any Contribution intentionally submitted for inclusion in the Work + by You to the Licensor shall be under the terms and conditions of + this License, without any additional terms or conditions. + Notwithstanding the above, nothing herein shall supersede or modify + the terms of any separate license agreement you may have executed + with Licensor regarding such Contributions. + + 6. Trademarks. This License does not grant permission to use the trade + names, trademarks, service marks, or product names of the Licensor, + except as required for reasonable and customary use in describing the + origin of the Work and reproducing the content of the NOTICE file. + + 7. Disclaimer of Warranty. Unless required by applicable law or + agreed to in writing, Licensor provides the Work (and each + Contributor provides its Contributions) on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + implied, including, without limitation, any warranties or conditions + of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A + PARTICULAR PURPOSE. You are solely responsible for determining the + appropriateness of using or redistributing the Work and assume any + risks associated with Your exercise of permissions under this License. + + 8. Limitation of Liability. In no event and under no legal theory, + whether in tort (including negligence), contract, or otherwise, + unless required by applicable law (such as deliberate and grossly + negligent acts) or agreed to in writing, shall any Contributor be + liable to You for damages, including any direct, indirect, special, + incidental, or consequential damages of any character arising as a + result of this License or out of the use or inability to use the + Work (including but not limited to damages for loss of goodwill, + work stoppage, computer failure or malfunction, or any and all + other commercial damages or losses), even if such Contributor + has been advised of the possibility of such damages. + + 9. Accepting Warranty or Additional Liability. While redistributing + the Work or Derivative Works thereof, You may choose to offer, + and charge a fee for, acceptance of support, warranty, indemnity, + or other liability obligations and/or rights consistent with this + License. However, in accepting such obligations, You may act only + on Your own behalf and on Your sole responsibility, not on behalf + of any other Contributor, and only if You agree to indemnify, + defend, and hold each Contributor harmless for any liability + incurred by, or claims asserted against, such Contributor by reason + of your accepting any such warranty or additional liability. + + END OF TERMS AND CONDITIONS + + APPENDIX: How to apply the Apache License to your work. + + To apply the Apache License to your work, attach the following + boilerplate notice, with the fields enclosed by brackets "[]" + replaced with your own identifying information. (Don't include + the brackets!) The text should be enclosed in the appropriate + comment syntax for the file format. We also recommend that a + file or class name and description of purpose be included on the + same "printed page" as the copyright notice for easier + identification within third-party archives. + + Copyright [yyyy] [name of copyright owner] + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. diff --git a/README.md b/README.md index 3fb8f3f..0050b8c 100644 --- a/README.md +++ b/README.md @@ -1,73 +1,19 @@ -# Kungfu Panda -**Kungfu Panda** is a library for register python pandas UDFs in Spark SQL. +# Bamboo -# Quick Start +Spark运行MLFlow模型的缓存组件 -1. download project. -``` -git clone https://github.com/cfmcgrady/kungfu-panda.git -``` - -2. install python environment by conda. -``` -conda env create -f path/to/conda.yaml -p /tmp/kungfu-panda -``` +# 环境变量依赖 -3. train a Kmean classify model with mlflow. -``` -/tmp/kungfu-panda/bin/python path/to/train.py -``` +构建docker镜像和部署服务都需要依赖相关的环境变量。 -4. register model. -```scala - val spark = SparkSession - .builder() - .appName("kungfu panda example") - .master("local[4]") - .getOrCreate() +| 环境变量 | 说明 | +| --- | --- | +| AWS_ACCESS_KEY_ID | minio访问id，需要有mlflow的bucket权限| +| AWS_SECRET_ACCESS_KEY | minio访问key，需要有mlflow的bucket权限| +| MLFLOW_S3_ENDPOINT_URL | minio地址，外部测试时候使用http://minio.k8s.uc.host.dxy，部署到k8s的时候需要内部域名 | +| BAMBOO_CACHE_DIR | 缓存根路径，默认为/tmp/cache | - val python = "/tmp/kungfu-panda/bin/python" - val artifactRoot = "." - // find run id with mlflow. - val runid = "9c6c59d0f57f40dfbbded01816896687" - val pythonExec = Option(python) - PandasFunctionManager.registerMLFlowPythonUDF( - spark, "test", - returnType = Option(IntegerType), - artifactRoot = Option(artifactRoot), - runId = runid, - driverPythonExec = pythonExec, - driverPythonVer = None, - pythonExec = pythonExec, - pythonVer = None) - spark.sql( - """ - |select test(x, y) from ( - |select 1 as x, 1 as y - |) - |""".stripMargin) - .show() +build docker image: +```shell +docker build -t bamboo -f dev/bamboo/Dockerfile . ``` - -# Register Function With Spark SQL - -1. add parser extensions when we create `SparkSession` -```scala -val spark = SparkSession - .builder() - .appName("panda sql example") - .master("local[4]") - .withExtensions(CreateFunctionParser.extBuilder) - .getOrCreate() -``` - -2. register mlflow function. -```sql -CREATE FUNCTION `test` AS '${runid}' USING `type` 'mlflow', `returns` 'integer', `artifactRoot` '${artifactRoot}', `pythonExec` '${python}' -``` - -visit [PandaSqlExample](./examples/local/src/main/scala/org/panda/example/local/PandaSqlExample.scala) for full example. - -# Run On Yarn Cluster - -// todo diff --git a/assembly/pom.xml b/assembly/pom.xml index 426b2d0..d7bfc78 100644 --- a/assembly/pom.xml +++ b/assembly/pom.xml @@ -13,7 +13,9 @@ pom - ${project.parent.basedir}/libs + + tgz + * @@ -25,11 +27,10 @@ 3.0.0 - ${project.basedir}/src/main/assembly/assembly.xml - + ${project.basedir}/src/main/assembly/assembly.xml - ${project.basedir} + ${project.parent.basedir} diff --git a/assembly/src/main/assembly/assembly.xml b/assembly/src/main/assembly/assembly.xml index 3774a1d..7bd06e1 100644 --- a/assembly/src/main/assembly/assembly.xml +++ b/assembly/src/main/assembly/assembly.xml @@ -1,16 +1,33 @@ bin-${project.version} - tgz + ${assembly.format} - - - ${project.parent.basedir}/conf - conf + + + + true + - **/* + + org.panda:${assembly.target.module} - + + libs + ${assembly.unpack} + + + + + + + + + + + + + ${project.parent.basedir} . @@ -20,8 +37,8 @@ - ${project.parent.basedir}/libs - libs + ${project.parent.basedir}/conf + conf **/* diff --git a/bamboo/pom.xml b/bamboo/pom.xml index abbb032..edbbff0 100644 --- a/bamboo/pom.xml +++ b/bamboo/pom.xml @@ -14,6 +14,7 @@ ${project.parent.basedir}/libs + 2.0.2.RELEASE @@ -30,12 +31,17 @@ org.springframework.boot spring-boot-starter-jetty - 2.0.2.RELEASE + ${spring-boot.version} org.springframework.boot spring-boot-starter-web - 2.0.2.RELEASE + ${spring-boot.version} + + + org.springframework.boot + spring-boot-starter-actuator + ${spring-boot.version} io.springfox @@ -50,13 +56,23 @@ com.fasterxml.jackson.module jackson-module-scala_${scala.binary.version} - 2.9.9 com.fasterxml.jackson.core jackson-databind - 2.9.9.3 - + + + single-jar + + + + org.apache.maven.plugins + maven-shade-plugin + + + + + diff --git a/bamboo/src/main/scala/org/panda/bamboo/service/controller/CondaController.scala b/bamboo/src/main/scala/org/panda/bamboo/service/controller/CondaController.scala index c291578..ae7c02c 100644 --- a/bamboo/src/main/scala/org/panda/bamboo/service/controller/CondaController.scala +++ b/bamboo/src/main/scala/org/panda/bamboo/service/controller/CondaController.scala @@ -1,15 +1,20 @@ package org.panda.bamboo.service.controller -import java.util.Base64 +import java.nio.file.{Files, Paths} +import java.util.{Base64, HashMap => JMap} import scala.util.control.NonFatal import org.apache.catalina.servlet4preview.http.HttpServletRequest import org.apache.spark.panda.utils.Conda -import org.panda.bamboo.util.{CacheKey, CacheManager} +import org.panda.bamboo.util.{CacheKey, CacheManager, PythonEnvironmentResolvedPath} +import org.panda.Config +import org.slf4j.LoggerFactory import org.springframework.core.io.{Resource, UrlResource} import org.springframework.http.{HttpHeaders, MediaType, ResponseEntity} -import org.springframework.web.bind.annotation.{PathVariable, RequestBody, RequestMapping, RequestMethod, RequestParam, RestController} +import org.springframework.web.bind.annotation.{PathVariable, PostMapping, RequestBody, RequestMapping, RequestMethod, RequestParam, RestController} +import org.springframework.web.multipart.MultipartFile +import org.springframework.web.servlet.mvc.support.RedirectAttributes /** * @time 2019-08-30 10:23 @@ -19,6 +24,8 @@ import org.springframework.web.bind.annotation.{PathVariable, RequestBody, Reque @RequestMapping(value = Array("/api/v1/conda")) class CondaController { + private val logger = LoggerFactory.getLogger(getClass.getCanonicalName) + /** * post method for createAndGet. * @param yaml @@ -32,7 +39,7 @@ class CondaController { println(yaml) // scalastyle:on val name = CacheManager.get(key(yaml)) - val resource = new UrlResource(CacheManager.getFileByName(name).toUri) + val resource = new UrlResource(PythonEnvironmentResolvedPath.compressFilePath(name).toUri) var contentType = "" @@ -102,7 +109,7 @@ class CondaController { def directGet(@PathVariable filename: String, request: HttpServletRequest): ResponseEntity[Resource] = { - val file = s"file:///tmp/cache/${filename}/${filename}.tgz" + val file = s"file://${Config.CACHE_ROOT_DIR}/${filename}/${filename}.tgz" val resource = new UrlResource(file) var contentType = "" try { @@ -133,6 +140,38 @@ class CondaController { Response(data = Map("result" -> Base64.getEncoder.encodeToString(yaml.getBytes("utf-8")))) } + @RequestMapping(value = Array("/admin/remove/{md5}"), method = Array(RequestMethod.DELETE)) + def remove(@PathVariable md5: String): Response = { + try { + CacheManager.remove(CacheKey(md5, new JMap[String, Object]())) + Response() + } catch { + case t: Throwable => + logger.error(s"catch an exception when remove environment $md5", t) + Response(stat = false, message = t.getMessage) + } + } + + @PostMapping(value = Array("admin/upload")) + def manullyUpload(@RequestParam("file") file: MultipartFile, + redirectAttributes: RedirectAttributes): Unit = { + if (file.isEmpty()) { + redirectAttributes.addFlashAttribute("message", "Please select a file to upload") + return "redirect:uploadStatus" + } + try { + // Get the file and save it somewhere + val bytes = file.getBytes() + val path = Paths.get("/tmp/dd" + file.getOriginalFilename()) + Files.write(path, bytes) + redirectAttributes.addFlashAttribute("message", + "You successfully uploaded '" + file.getOriginalFilename() + "'") + } catch { + case e => e.printStackTrace() + } + return "redirect:/uploadStatus"; + } + private def key(yaml: String): CacheKey = { val ymap = Conda.normalize(yaml) CacheKey(ymap.getOrDefault("name", "").asInstanceOf[String], ymap) diff --git a/bamboo/src/main/scala/org/panda/bamboo/service/controller/MLFlowArtifactController.scala b/bamboo/src/main/scala/org/panda/bamboo/service/controller/MLFlowArtifactController.scala index 952736f..ef0d18e 100644 --- a/bamboo/src/main/scala/org/panda/bamboo/service/controller/MLFlowArtifactController.scala +++ b/bamboo/src/main/scala/org/panda/bamboo/service/controller/MLFlowArtifactController.scala @@ -2,8 +2,8 @@ package org.panda.bamboo.service.controller import java.net.URI import java.nio.file.Paths - import javax.servlet.http.HttpServletRequest + import org.panda.bamboo.util.{CacheManager, MLFlowRunCacheKey} import org.springframework.core.io.{Resource, UrlResource} import org.springframework.http.{HttpHeaders, MediaType, ResponseEntity} @@ -48,7 +48,7 @@ class MLFlowArtifactController { .body(resource) } - @RequestMapping(value = Array("/admin/remove/{runid}"), method = Array(RequestMethod.GET, RequestMethod.POST)) + @RequestMapping(value = Array("/admin/remove/{runid}"), method = Array(RequestMethod.DELETE)) def remove(@PathVariable runid: String): Response = { try { CacheManager.remove(key(runid)) diff --git a/bamboo/src/main/scala/org/panda/bamboo/util/CacheEntity.scala b/bamboo/src/main/scala/org/panda/bamboo/util/CacheEntity.scala index 0df60ee..3d2a835 100644 --- a/bamboo/src/main/scala/org/panda/bamboo/util/CacheEntity.scala +++ b/bamboo/src/main/scala/org/panda/bamboo/util/CacheEntity.scala @@ -8,8 +8,10 @@ import java.util.concurrent.atomic.AtomicBoolean import java.util.concurrent.locks.ReentrantReadWriteLock import org.apache.commons.compress.archivers.tar.TarArchiveEntry +import org.apache.commons.io.FileUtils import org.apache.commons.logging.LogFactory -import org.apache.spark.panda.utils.{CompressUtil, Conda, SFTPUtil, Util} +import org.apache.spark.panda.utils.{CompressUtil, Conda, MLFlowMinioUtilImpl, MLFlowUtil, SFTPUtil, Util} +import org.panda.Config /** * @time 2019-09-12 16:48 @@ -18,7 +20,7 @@ import org.apache.spark.panda.utils.{CompressUtil, Conda, SFTPUtil, Util} trait CacheEntity[T] { private val _lock = new ReentrantReadWriteLock() - private val _cacheVaild: AtomicBoolean = new AtomicBoolean(false) + private val _cacheValid: AtomicBoolean = new AtomicBoolean(false) protected def write: Unit @@ -30,7 +32,9 @@ trait CacheEntity[T] { _lock.writeLock().lock() try { delete - _cacheVaild.set(false) + _cacheValid.set(false) + } catch { + case t: Throwable => throw t } finally { _lock.writeLock().unlock() } @@ -39,15 +43,15 @@ trait CacheEntity[T] { def get(): T = { _lock.readLock().lock() - if (!_cacheVaild.get()) { + if (!_cacheValid.get()) { _lock.readLock().unlock() _lock.writeLock().lock() try { - if (!_cacheVaild.get()) { + if (!_cacheValid.get()) { // do package download - // TODO:(fchen) throws execption when we has downloaded fail. + // TODO:(fchen) throws exception when we has downloaded fail. write - _cacheVaild.set(true) + _cacheValid.set(true) } _lock.readLock().lock() } finally { @@ -67,15 +71,18 @@ class PythonEnvironmentCacheEntity ( name: String, configuration: JMap[String, Object]) extends CacheEntity[String] { + import CacheManager._ + val logger = LogFactory.getLog(this.getClass) + val resolvedEnvRootPath: Path = PythonEnvironmentResolvedPath.resolveEnvPath(name) + private def downloadAndPackage(): Unit = { - import CacheManager._ - if (!(Paths.get(basePath, Array(name): _*).toFile.exists() && - Paths.get(basePath, Array(name, s"${name}.tgz"): _*).toFile.exists())) { + if (!resolvedEnvRootPath.toFile.exists() && + !PythonEnvironmentResolvedPath.compressFilePath(name).toFile.exists()) { logger.info("env not found, begin download from internet.") // the environment has never been download before, so we download this package now. - val envpath = Conda.createEnv(name, configuration, basePath + File.separator + name) + val envpath = Conda.createEnv(name, configuration, resolvedEnvRootPath.toString) // make python command executable. val makeExecutable = { @@ -95,7 +102,10 @@ class PythonEnvironmentCacheEntity ( override protected def read: String = name - override def delete: Unit = throw new UnsupportedOperationException("") + override def delete: Unit = { + FileUtils.deleteDirectory(PythonEnvironmentResolvedPath.resolveEnvPath(name).toFile) +// Util.recursiveDeleteFile(envRootPath) + } /** * . @@ -106,53 +116,66 @@ class PythonEnvironmentCacheEntity ( */ } -class MLFlowRunCacheEntity(runid: String) extends CacheEntity[String] { +class MLFlowRunCacheEntity(runid: String) extends CacheEntity[String] + with MLFlowUtil + with ResolvedPath { + private val logger = LogFactory.getLog(getClass) -// private lazy val ARTIFACT_ROOT = new URI(sys.env.getOrElse("MLFLOW_ARTIFACT_ROOT", throw new RuntimeException(""))) - private lazy val ARTIFACT_ROOT = resolveURI("/Users/fchen/Project/python/mlflow-study/mlruns") + private lazy val MLFLOW_TRACKING_URI = sys.env.getOrElse("MLFLOW_TRACKING_URI", + throw new IllegalArgumentException("please set MLFLOW_TRACKING_URI environment variable.") + ) + + override def mlflowTrackingUri: String = MLFLOW_TRACKING_URI + + logger.info(s"service start with MLFLOW_TRACKING_URI = ${mlflowTrackingUri}") - private lazy val BASE_PATH = "/tmp/runs" + private val resolvedRunPath = resolveRunPath(runid) + + private val resolvedCompressionPath = compressFilePath(runid) override protected def write: Unit = { // make sure this run was not downloaded before. - if (new File(compressFilePath).getParentFile.exists()) { + if (new File(compressFilePath(runid)).getParentFile.exists()) { return } - logger.info(s"start to download run ${runid} from artifact ${ARTIFACT_ROOT}") - val path = ARTIFACT_ROOT.getScheme.toLowerCase match { + artifactUri(runid) match { + case Right(uri) => + logger.info(s"start to download run ${runid} from artifact ${uri}") + downloadAndCompress(new URI(uri)) + case Left(e) => + throw e + } + } + + def downloadAndCompress(uri: URI): Unit = { + uri.getScheme.toLowerCase match { case "file" => - Util.getArtifactByRunId(ARTIFACT_ROOT.getPath, runid) + val path = Util.getArtifactByRunId(uri.getPath, runid) + CompressUtil.tar2(path, resolvedCompressionPath) case "sftp" => // TODO:(fchen) we should look the remote path from mlflow tracking server api. - val remotePath = ARTIFACT_ROOT.getPath + "/0/" + runid - SFTPUtil.download(ARTIFACT_ROOT.getHost, remotePath, "/tmp/runs") - s"${BASE_PATH}/$runid" + val remotePath = uri.getPath + "/0/" + runid + // make sure the `BASE_PATH` exist. otherwise we should create the directory manually. + Util.mkdir(resolvedRunPath) + + SFTPUtil.download(uri.getHost, remotePath, resolvedRunPath) + CompressUtil.tar2(resolvedRunPath, resolvedCompressionPath) + case "s3" => + // make sure the `BASE_PATH` exist. otherwise we should create the directory manually. + Util.mkdir(resolvedRunPath) + MLFlowMinioUtilImpl.downloadAsZip(uri.getHost, uri.getPath, resolvedCompressionPath) case _ => throw new UnsupportedOperationException() } - CompressUtil.tar(path, compressFilePath) } - def compressFilePath: String = { - s"${BASE_PATH}/${runid}/${runid}.tgz" - } - - override protected def read: String = compressFilePath + override protected def read: String = compressFilePath(runid) override def delete: Unit = { - try { - Util.recursiveListFiles(Paths.get(s"${BASE_PATH}/${runid}").toFile) - .foreach(f => { - Files.deleteIfExists(f.toPath) - }) - Files.deleteIfExists(Paths.get(s"${BASE_PATH}/${runid}")) - } catch { - case e: IOException => - logger.info(s"remove run $runid failed!", e) - } + Util.recursiveDeleteFile(resolvedRunPath) } private def resolveURI(path: String): URI = { @@ -173,5 +196,34 @@ class MLFlowRunCacheEntity(runid: String) extends CacheEntity[String] { } new File(path).getAbsoluteFile().toURI() } +} + +trait ResolvedPath { + self: MLFlowRunCacheEntity => + private lazy val BASE_PATH = sys.env.getOrElse("panda.cache.dir", s"${Config.CACHE_ROOT_DIR}/panda/runs") + + /** + * the root cache path of this run. + */ + protected val resolveRunPath = (runid: String) => s"${BASE_PATH}/${runid}" + + /** + * the compressed file path of this run. + */ + val compressFilePath = { + runid: String => s"${resolveRunPath(runid)}/${runid}.tgz" + } + +} + +object PythonEnvironmentResolvedPath { + // // TODO:(fchen) generate base path with server info(hostname: port). + // // so that we can deploy multi server on the same host. + // val basePath = Config.CACHE_ROOT_DIR + private lazy val BASE_PATH = s"${Config.CACHE_ROOT_DIR}/conda" + + val resolveEnvPath = (name: String) => Paths.get(BASE_PATH, Array(name): _*) + + val compressFilePath = (name: String) => Paths.get(BASE_PATH, Array(name, s"${name}.tgz"): _*) } diff --git a/bamboo/src/main/scala/org/panda/bamboo/util/CacheManager.scala b/bamboo/src/main/scala/org/panda/bamboo/util/CacheManager.scala index e3b7f76..9fa6371 100644 --- a/bamboo/src/main/scala/org/panda/bamboo/util/CacheManager.scala +++ b/bamboo/src/main/scala/org/panda/bamboo/util/CacheManager.scala @@ -2,10 +2,12 @@ package org.panda.bamboo.util import java.nio.file.{Path, Paths} import java.util.{Map => JMap} +import java.util.concurrent.locks.ReentrantLock import com.google.common.cache.{CacheBuilder, CacheLoader} import org.apache.juli.logging.LogFactory import org.apache.spark.panda.utils.Conda +import org.panda.Config /** * @time 2019-08-29 14:33 @@ -15,28 +17,10 @@ object CacheManager { private val logger = LogFactory.getLog(getClass) - private lazy val _pythonEnvCache = CacheBuilder.newBuilder() - .maximumSize(100) - .build( - new CacheLoader[CacheKey, PythonEnvironmentCacheEntity] { - override def load(k: CacheKey): PythonEnvironmentCacheEntity = { - new PythonEnvironmentCacheEntity(k.name, k.conf) - } - } - ) - - private lazy val _mlflowRunCache = CacheBuilder.newBuilder() - .maximumSize(100) - .build( - new CacheLoader[MLFlowRunCacheKey, MLFlowRunCacheEntity] { - override def load(k: MLFlowRunCacheKey): MLFlowRunCacheEntity = { - new MLFlowRunCacheEntity(k.runid) - } - } - ) - + private val _lock = new ReentrantLock() + // todo:(fchen) 基于文件锁来实现 private lazy val _cache = CacheBuilder.newBuilder() - .maximumSize(1000) + .maximumSize(100000) .build( new CacheLoader[Key, CacheEntity[String]] { override def load(key: Key): CacheEntity[String] = { @@ -51,7 +35,9 @@ object CacheManager { ) def get: (Key) => String = { - key => _cache.get(key).get() + withLock { + key => _cache.get(key).get() + } // case k: CacheKey => // _pythonEnvCache.get(k).get() // case k: MLFlowRunCacheKey => @@ -59,46 +45,21 @@ object CacheManager { } def remove: (Key) => Unit = { - key => - logger.info(s"start to remove ${key}") - _cache.get(key).remove - } - - /** - * get the tar archive file path by environment name. - */ - def getFileByName(name: String): Path = { - Paths.get(basePath, Array(name, s"${name}.tgz"): _*) + withLock { + key => + logger.info(s"start to remove ${key}") + _cache.get(key).remove + _cache.invalidate(key) + } } - // TODO:(fchen) generate base path with server info(hostname: port). - // so that we can deploy multi server on the same host. - val basePath = "/tmp/cache" - - def main(args: Array[String]): Unit = { - val yaml = - """ - |channels: - | - http://nexus.k8s.uc.host.dxy/repository/anaconda/pkgs/main/ - | - http://nexus.k8s.uc.host.dxy/repository/anaconda/pkgs/free/ - | - http://nexus.k8s.uc.host.dxy/repository/anaconda/pkgs/r/ - | - http://nexus.k8s.uc.host.dxy/repository/anaconda/pkgs/pro/ - | - http://nexus.k8s.uc.host.dxy/repository/anaconda/pkgs/msys2/ - |dependencies: - |- python=3.6.0 - |- numpy - |name: conda-test - """.stripMargin - - (1 to 10).foreach(i => { - new Thread(new Runnable { - override def run(): Unit = { - val ymap = Conda.normalize(yaml) - val k = CacheKey(ymap.getOrDefault("name", "").asInstanceOf[String], ymap) - get(k) - } - }).start() - }) + private def withLock[T](f: T): T = { + _lock.lock() + try { + f + } finally { + _lock.unlock() + } } } diff --git a/bootstrap.sh b/bootstrap.sh new file mode 100755 index 0000000..4b39352 --- /dev/null +++ b/bootstrap.sh @@ -0,0 +1,156 @@ +#!/bin/bash + +APP_NAME=Merlion +APP_VERSION=1.0 +JAR_NAME=${APP_NAME}-${APP_VERSION}.jar +#export JAVA_HOME=/opt/java8/ + +# 启动时间版本 + +DATE_VERSION=$(date +%Y%m%d%H%M%S) + +# 生产配置 +BIN_DIR=/usr/local/release/${APP_NAME}/ +PID_FILE=/tmp/${APP_NAME}.pid +LOG_FILE=/tmp/${APP_NAME}.${DATE_VERSION}.log +MAIN_CLASS=com.dxy.data.merlion.service.DevApplication + +export GLOG_v=1 +export GLOG_log_dir=/tmp/a +# export LIBPROCESS_IP=10.25.26.135 + +# 自定义配置 +export MESOS_NATIVE_JAVA_LIBRARY=/usr/local/lib/libmesos.so + +if [ -z "${DXY_PROJECT_HOME}" ]; then + export DXY_PROJECT_HOME="$(cd "`dirname "$0"`"/.; pwd)" +fi + +JAVA_OPTS="-Dserver.port=8100 -Dfile.encoding=UTF-8 -Dlog4j.configuration=file://${DXY_PROJECT_HOME}/conf/log4j.properties -Dlog4j.configurationFile=file://${DXY_PROJECT_HOME}/conf/log4j2.xml" + +JAVA_OPTS=${JAVA_OPTS}" -Dsentry.dsn=https://87727747fd244d61a8dda6339cb1b657:48ed7ab92a774313bca9e326653c4831@sentry.k8s.uc.host.dxy/82" + + +JAVA_OPTS=${JAVA_OPTS}" -Xms8g -Xmx8g \ + -XX:ParallelGCThreads=8 \ + -XX:SurvivorRatio=1 \ + -XX:LargePageSizeInBytes=128M \ + -XX:MaxNewSize=1g \ + -XX:CMSInitiatingOccupancyFraction=80 \ + -XX:+UseCMSCompactAtFullCollection \ + -XX:CMSFullGCsBeforeCompaction=0 \ + -XX:-UseGCOverheadLimit \ + -XX:MaxTenuringThreshold=5 \ + -XX:GCTimeRatio=19 \ + -XX:+UseConcMarkSweepGC \ + -XX:+UseParNewGC \ + -XX:+PrintGCDetails \ + -XX:+PrintGCTimeStamps \ + -XX:+HeapDumpOnOutOfMemoryError \ + -XX:HeapDumpPath=/tmp/${APP_NAME}-${MODULE}.dump \ + -Xloggc:/tmp/${APP_NAME}-${MODULE}-gc.$DATE_VERSION.log" +JARS=$(echo ${DXY_PROJECT_HOME}/libs/*.jar | tr ' ' ':') + +function status() { + echo "$APP_NAME Status" + if [ -s ${PID_FILE} ]; then + ps h -fp $(cat ${PID_FILE}) + fi +} + +function common_run() { + $JAVA_HOME/bin/java -cp $JARS "$@" +} + +function package() { + cd $DXY_PROJECT_HOME + echo "Remove ${APP_NAME}-bin-${APP_VERSION}.tgz..." + rm ${APP_NAME}-bin-${APP_VERSION}.tgz + echo "Package App $APP_NAME-$APP_VERSION" + mvn clean package -DskipTests "$@" + mv $DXY_PROJECT_HOME/assembly/${APP_NAME}-bin-${APP_VERSION}.tgz $DXY_PROJECT_HOME +} + +function usage() { +cat << EOF + Usage: ./bootstrap.sh package +EOF +} + +function echo_build_properties() { + echo version=$APP_VERSION + echo user=$USER + echo revision=$(git rev-parse HEAD) + echo branch=$(git rev-parse --abbrev-ref HEAD) + echo date=$(date +"%Y/%m/%d %H:%M:%S") + echo url=$(git config --get remote.origin.url) +} + +function build_info() { + echo_build_properties $2 > $DXY_PROJECT_HOME/INFO +} + +function start_frontend() { + $JAVA_HOME/bin/java $JAVA_OPTS -cp $JARS $MAIN_CLASS +} + +function start() { + echo "Start App $APP_NAME" + if [ -s ${PID_FILE} ]; then + r=`ps h -fp $(cat ${PID_FILE})` + fi + if [ "$r" == "" ]; then + nohup $JAVA_HOME/bin/java $JAVA_OPTS -cp $JARS $MAIN_CLASS > ${LOG_FILE} 2>&1 & echo $! > ${PID_FILE} + echo "${APP_NAME} log file ${LOG_FILE}" + else + echo "${APP_NAME} already running..." + fi +} + +function stop() { + echo "Stop App $APP_NAME" + if [ -s ${PID_FILE} ]; then + echo "stopping ${APP_NAME}: $(cat ${PID_FILE})" + kill -9 $(cat ${PID_FILE}) + rm -f ${PID_FILE} + else + echo "pid file not found" + exit 1 + fi +} + +function restart() { + stop + start +} + +case "$1:$2:$3" in + package:*) + package "${@:2}" + ;; + run:*:*) + common_run "${@:2}" + ;; + build_info:*:*|bi:*:*) + build_info + ;; + start_frontend:*) + start_frontend + ;; + start:*) + start + ;; + stop:*) + stop + ;; + restart:*) + restart + ;; + h|help) + usage + ;; + *) + usage + exit 0 +esac + diff --git a/common/pom.xml b/common/pom.xml index a231c62..187167b 100644 --- a/common/pom.xml +++ b/common/pom.xml @@ -20,7 +20,7 @@ org.apache.commons commons-compress - 1.18 + 1.19 org.yaml @@ -37,6 +37,19 @@ sshj 0.27.0 + + commons-io + commons-io + 2.6 + + + org.mlflow + mlflow-client + + + io.minio + minio + diff --git a/common/src/main/java/org/apache/spark/panda/utils/GZIPUtil.java b/common/src/main/java/org/apache/spark/panda/utils/GZIPUtil.java index 8c455d3..0b42e69 100644 --- a/common/src/main/java/org/apache/spark/panda/utils/GZIPUtil.java +++ b/common/src/main/java/org/apache/spark/panda/utils/GZIPUtil.java @@ -37,6 +37,7 @@ public static void createTarArchive(String parentDir, String outFile){ e.printStackTrace(); }finally{ try { + tarArchive.finish(); tarArchive.close(); } catch (IOException e) { // TODO Auto-generated catch block @@ -54,13 +55,16 @@ public static void addToArchive(String filePath, String parent, TarArchiveOutput // add tar ArchiveEntry // tarEntry.setMode(755) - tarArchive.putArchiveEntry(new TarArchiveEntry(file, entryName)); + TarArchiveEntry tae = new TarArchiveEntry(file, entryName); + tae.setSize(file.length()); + tarArchive.putArchiveEntry(tae); if(file.isFile()) { -// FileInputStream fis = new FileInputStream(file); -// BufferedInputStream bis = new BufferedInputStream(fis); + FileInputStream fis = new FileInputStream(file); + BufferedInputStream bis = new BufferedInputStream(fis); // // Write file content to archive // IOUtils.copy(bis, tarArchive); - Files.copy(file.toPath(), tarArchive); + org.apache.commons.io.IOUtils.copyLarge(bis, tarArchive); +// Files.copy(file.toPath(), tarArchive); tarArchive.closeArchiveEntry(); // bis.close(); }else if(file.isDirectory()) { diff --git a/common/src/main/scala/org/apache/spark/panda/utils/CompressUtil.scala b/common/src/main/scala/org/apache/spark/panda/utils/CompressUtil.scala index 96ae94b..23e7e47 100644 --- a/common/src/main/scala/org/apache/spark/panda/utils/CompressUtil.scala +++ b/common/src/main/scala/org/apache/spark/panda/utils/CompressUtil.scala @@ -15,7 +15,6 @@ import org.apache.commons.compress.compressors.gzip.GzipCompressorOutputStream * @author fchen */ object CompressUtil { - def zip(sourceDirectory: String, targetZipFile: String): Unit = { val p = Files.createFile(Paths.get(targetZipFile)) val zs = new ZipOutputStream(Files.newOutputStream(p)) @@ -58,12 +57,13 @@ object CompressUtil { case file => val path = file.toPath val tarEntry = new TarArchiveEntry(path.toFile, sourceDirectoryPath.getParent.relativize(path).toString()) - action(path, tarEntry) - taos.putArchiveEntry(tarEntry) -// val in = new FileInputStream(path.toFile) - Files.copy(path, taos) -// IOUtils.copy(in, taos) - taos.closeArchiveEntry() + try { + action(path, tarEntry) + taos.putArchiveEntry(tarEntry) + Files.copy(path, taos) + } finally { + taos.closeArchiveEntry() + } } } catch { case e: Exception => @@ -81,12 +81,6 @@ object CompressUtil { GZIPUtil.createTarArchive(sourceDirectory, targetTarFile) } - def tar3(): Unit = { - Process( - "tar czf " - ) - } - @throws(classOf[IOException]) def unzip(sourceZipFile: String, uncompressedDirectory: String, diff --git a/common/src/main/scala/org/apache/spark/panda/utils/Conda.scala b/common/src/main/scala/org/apache/spark/panda/utils/Conda.scala index d099f32..aceaf72 100644 --- a/common/src/main/scala/org/apache/spark/panda/utils/Conda.scala +++ b/common/src/main/scala/org/apache/spark/panda/utils/Conda.scala @@ -8,6 +8,7 @@ import scala.collection.JavaConverters._ import scala.collection.mutable.Buffer import scala.sys.process.{Process, ProcessLogger} +import org.slf4j.LoggerFactory import org.yaml.snakeyaml.Yaml /** @@ -15,6 +16,13 @@ import org.yaml.snakeyaml.Yaml * @author fchen */ object Conda { + + val logger = LoggerFactory.getLogger(getClass.getCanonicalName) + + private val CONDA_COMMAND = sys.env.getOrElse("CONDA_PATH", "conda") + + logger.info(s"using conda path = ${CONDA_COMMAND}.") + def createEnv(name: String, yaml: JMap[String, Object], basePath: String): Path = { @@ -30,31 +38,39 @@ object Conda { } try { - val cmd = s"conda env create -f ${yamlPath.toString} -p ${envPath}" + val cmd = s"${CONDA_COMMAND} env create -f ${yamlPath.toString} -p ${envPath}" + logger.info(s"running command [ ${cmd} ].") // scalastyle:off println - val logger = ProcessLogger(println, println) + val processLogger = ProcessLogger(logger.info, logger.info) // scalastyle:on Process( cmd - ).!!(logger) + ).!!(processLogger) } catch { case e: RuntimeException => e.printStackTrace() } + logger.info(s"finished create environment $name.") envPath } - def normalize(configurations: String): JMap[String, Object] = { + def normalize(configurations: String, withPyarrow: Boolean = true): JMap[String, Object] = { val yaml = new Yaml() val info = yaml.load[JMap[String, Object]](configurations) val dependencies = info.get("dependencies") + if (withPyarrow) { + // add pyarrow dependency for pyspark runtime. + addPyarrow(dependencies.asInstanceOf[JArrayList[Object]]) + } + val (dep, pip) = extract(dependencies.asInstanceOf[JArrayList[Object]].asScala) + val nname = generateEnvironmentName(dep, pip) + info.put("name", nname) - val total = dep ++ pip + // remove user define channels. + info.remove("channels") - val nname = Util.stringToMD5(total.sortBy(x => x).mkString(",")) - info.put("name", nname) info } @@ -87,6 +103,7 @@ object Conda { } yaml.dump(conf, new FileWriter(filePath.toFile)) + logger.info(s"write [ ${conf.asScala.mkString(",")} ] into [ ${filePath} ] success.") } /** @@ -98,8 +115,36 @@ object Conda { conf } - def addPyarrow(configurations: JMap[String, Object]): Unit = { -// pyarrow==0.12.1 + def addPyarrow(pip: Buffer[String]): Buffer[String] = { + pip.filter(!_.startsWith("pyarrow")) += PYARROW + } + + def addPyarrow(dependencies: JArrayList[Object]): Unit = { + dependencies.asScala + .collect { case map: java.util.LinkedHashMap[_, _] => map} + .headOption + .orElse { + val pip = new java.util.LinkedHashMap[Object, Object]() + pip.put("pip", new JArrayList[Object]()) + dependencies.add(pip) + Option(pip) + }.foreach(pip => { + pip.get("pip") + .asInstanceOf[JArrayList[Object]] + .add(PYARROW) + }) + } + + // todo: (fchen) read from system configurations. + val PYARROW = "pyarrow==0.12.1" + + private def generateEnvironmentName(dependencies: Buffer[String], pip: Buffer[String]): String = { + // We should sort the package first so that we can generate the unique id for the same Conda environment, + // in which the environment dependencies have a shuffled order. + Util.stringToMD5( + dependencies.sortBy(o => o).mkString("dependencies: [", ",", "]") + + pip.sortBy(o => o).mkString("pip: [", ",", "]") + ) } } diff --git a/common/src/main/scala/org/apache/spark/panda/utils/GZIPUtilV2.scala b/common/src/main/scala/org/apache/spark/panda/utils/GZIPUtilV2.scala new file mode 100644 index 0000000..a7dfe75 --- /dev/null +++ b/common/src/main/scala/org/apache/spark/panda/utils/GZIPUtilV2.scala @@ -0,0 +1,109 @@ +package org.apache.spark.panda.utils + +import java.io.{BufferedInputStream, BufferedOutputStream, File, FileInputStream, FileOutputStream, InputStream, IOException} +import java.util.zip.GZIPOutputStream + +import org.apache.commons.compress.archivers.tar.{TarArchiveEntry, TarArchiveOutputStream} +import org.apache.commons.io.IOUtils +import org.apache.commons.logging.LogFactory + +/** + * @time 2019/12/10 上午10:19 + * @author fchen + */ +object GZIPUtilV2 { + + private val logger = LogFactory.getLog(getClass) + + def createTarArchive(parentDir: String, outFile: String): Unit = { + var tarArchive: TarArchiveOutputStream = null + try { + val fos = new FileOutputStream(outFile) + val gzipOS = new GZIPOutputStream(new BufferedOutputStream(fos)) + tarArchive = new TarArchiveOutputStream(gzipOS) + tarArchive.setLongFileMode(TarArchiveOutputStream.LONGFILE_POSIX) + addToArchive(parentDir, "", tarArchive) + + } finally { + tarArchive.finish() + tarArchive.close() + } + } + + @throws(classOf[IOException]) + def addToArchive(filePath: String, + parent: String, + tarArchive: TarArchiveOutputStream): Unit = { + val file = new File(filePath); + // Create entry name relative to parent file path + // for the archived file + val entryName = parent + file.getName() + // scalastyle:off println + println("entryName " + entryName) + // scalastyle:on + // add tar ArchiveEntry + + // tarEntry.setMode(755) + val tae = new TarArchiveEntry(file, entryName) + tae.setSize(file.length()) + tarArchive.putArchiveEntry(tae) + if (file.isFile()) { + val fis = new FileInputStream(file) + val bis = new BufferedInputStream(fis) + // // Write file content to archive + // IOUtils.copy(bis, tarArchive); + org.apache.commons.io.IOUtils.copyLarge(bis, tarArchive) + // Files.copy(file.toPath(), tarArchive); + tarArchive.closeArchiveEntry() + // bis.close(); + } else if (file.isDirectory()) { + // no content to copy so close archive entry + tarArchive.closeArchiveEntry() + // if this directory contains more directories and files + // traverse and archive them + file.listFiles().foreach(f => addToArchive(f.getAbsolutePath, entryName + File.separator, tarArchive)) + } + } + + def streamCreateTarArchive(outFile: String)(f: TarArchiveOutputStream => Unit): Unit = { + var tarArchive: TarArchiveOutputStream = null + val fos = new FileOutputStream(outFile) + val gzipOS = new GZIPOutputStream(new BufferedOutputStream(fos)) + try { + tarArchive = new TarArchiveOutputStream(gzipOS) + tarArchive.setLongFileMode(TarArchiveOutputStream.LONGFILE_POSIX) + f(tarArchive) + } finally { + tarArchive.finish() + tarArchive.close() + gzipOS.flush() + gzipOS.close() + fos.flush() + fos.close() + } + } + + @throws(classOf[IOException]) + def streamAddToArchive(in: InputStream, + size: Long, + pathInArchive: String, + tarArchive: TarArchiveOutputStream): Unit = { + + logger.info(s"add entry ${pathInArchive} to archive.") + // add tar ArchiveEntry + + // tarEntry.setMode(755) + val tae = new TarArchiveEntry(pathInArchive) + tae.setSize(size) + tarArchive.putArchiveEntry(tae) + // // Write file content to archive + try { +// IOUtils.copy(in, tarArchive); + org.apache.commons.io.IOUtils.copyLarge(in, tarArchive) + } finally { + // Files.copy(file.toPath(), tarArchive); + tarArchive.closeArchiveEntry() + } + } + +} diff --git a/common/src/main/scala/org/apache/spark/panda/utils/MLFlow.scala b/common/src/main/scala/org/apache/spark/panda/utils/MLFlow.scala new file mode 100644 index 0000000..136e087 --- /dev/null +++ b/common/src/main/scala/org/apache/spark/panda/utils/MLFlow.scala @@ -0,0 +1,99 @@ +package org.apache.spark.panda.utils + +import java.util.{ArrayList => JArrayList, Map => JMap} + +import org.yaml.snakeyaml.Yaml + +/** + * @time 2019-09-16 15:47 + * @author fchen + */ +object MLFlow { + + def env(content: String): Unit = { +// val yaml = new Yaml() +// val mlmodel = yaml.load[JMap[String, Object]](content) +// val flavors = mlmodel.get("flavors").asInstanceOf[JMap[String, Object]] +// println(flavors.get("python_function").getClass) +// println(mlmodel.get("flavors").getClass) + val mlmodel = new MLmodelParser(content) + } + + def main(args: Array[String]): Unit = { + val yaml = + """ + |artifact_path: model + |flavors: + | python_function: + | data: model.pkl + | env: conda.yaml + | loader_module: mlflow.sklearn + | python_version: 3.6.9 + | sklearn: + | pickled_model: model.pkl + | serialization_format: cloudpickle + | sklearn_version: 0.19.1 + |run_id: 9c6c59d0f57f40dfbbded01816896687 + |utc_time_created: '2019-08-21 06:37:27.408296' + """.stripMargin + + val yaml2 = + """ + |flavors: + | python_function: + | artifacts: + | lgb_model: + | path: artifacts/lgb_model_path.pth + | uri: lgb_model_path.pth + | cloudpickle_version: 1.2.1 + | env: conda.yaml + | loader_module: mlflow.pyfunc.model + | python_model: python_model.pkl + | python_version: 3.6.0 + |utc_time_created: '2019-11-20 00:53:01.959671' + |""".stripMargin + + env(yaml) + } + +} + +/** + * a MLmodel file parser. + * here is a MLmodel example: + * -------------------------------------------------------- + * artifact_path: model + * flavors: + * python_function: + * data: model.pkl + * env: conda.yaml + * loader_module: mlflow.sklearn + * python_version: 3.6.9 + * sklearn: + * pickled_model: model.pkl + * serialization_format: cloudpickle + * sklearn_version: 0.19.1 + * run_id: 9c6c59d0f57f40dfbbded01816896687 + * utc_time_created: '2019-08-21 06:37:27.408296' + * -------------------------------------------------------- + * @param content + */ +class MLmodelParser(content: String) { + + val mlmodel = { + new Yaml().load[JMap[String, Object]](content) + } + + val artifactPath = typed[String](mlmodel.getOrDefault("artifact_path", "model")) + + val flavors = typed[JMap[String, Object]](mlmodel.get("flavors")) + + val pythonFunction = typed[JMap[String, Object]](flavors.get("python_function")) + + val env = typed[String](pythonFunction.get("env")) + + def typed[T](obj: Any): T = { + obj.asInstanceOf[T] + } + +} diff --git a/common/src/main/scala/org/apache/spark/panda/utils/MLFlowUtil.scala b/common/src/main/scala/org/apache/spark/panda/utils/MLFlowUtil.scala new file mode 100644 index 0000000..73c2d70 --- /dev/null +++ b/common/src/main/scala/org/apache/spark/panda/utils/MLFlowUtil.scala @@ -0,0 +1,56 @@ +package org.apache.spark.panda.utils + +import org.mlflow.tracking.{MlflowClient, MlflowHttpException} + +/** + * @time 2019/12/9 下午4:24 + * @author fchen + */ +trait MLFlowUtil { + def mlflowTrackingUri: String + protected val client = new MlflowClient(mlflowTrackingUri) + + def artifactUri(runId: String): Either[Throwable, String] = { + try { + Right(client.getRun(runId).getInfo.getArtifactUri) + } catch { +// case e: MlflowHttpException => +// e.printStackTrace() +// None + case t: Throwable => + Left(t) + } +// run.map(_.getInfo.getArtifactUri) + } +} + +object MLFlowUtilTest extends MLFlowUtil { + // http://localhost:5000/api/2.0/mlflow/experiments/list +// override def mlflowTrackingUri: String = "http://localhost:5000" +// override def mlflowTrackingUri: String = "http://192.168.205.91:9999" + override def mlflowTrackingUri: String = "http://mlflow.k8s.uc.host.dxy" + + def main(args: Array[String]): Unit = { +// import scala.collection.JavaConverters._ +// client.listExperiments().asScala.foreach(e => { +// println(e.getName) +// println(client.listRunInfos(e.getExperimentId).size()) +// // client.listArtifacts(e.getName).asScala.foreach(x => { +// // println(x.getPath) +// // }) +// client.listRunInfos(e.getExperimentId).asScala.foreach(x => { +// println(x.getArtifactUri) +// }) +// println(e.getArtifactLocation) +// }) +// +// println("-------") + val run = client.getRun("778825406cf44b02940a617b611c8384") + + run.getInfo + .getArtifactUri +// println(artifactUri("778825406cf44b02940a617b611c8384")) +// client.getExperiment("").getExperiment.get + + } +} diff --git a/common/src/main/scala/org/apache/spark/panda/utils/MinioUtil.scala b/common/src/main/scala/org/apache/spark/panda/utils/MinioUtil.scala new file mode 100644 index 0000000..8b05e0b --- /dev/null +++ b/common/src/main/scala/org/apache/spark/panda/utils/MinioUtil.scala @@ -0,0 +1,119 @@ +package org.apache.spark.panda.utils + +import java.net.{URI, URL} +import java.nio.file.Paths + +import scala.collection.JavaConverters._ + +import io.minio.MinioClient +import io.minio.messages.Item +import org.apache.commons.logging.LogFactory + +/** + * @time 2019/12/9 下午5:22 + * @author fchen + */ +trait MinioUtil { + + val logger = LogFactory.getLog(this.getClass) + + def serverInfo: S3ServerInfo + + val minioClient = + new MinioClient(serverInfo.endpoint, serverInfo.accessKeyId, serverInfo.secretAccessKey) + + def downloadAsZip(bucket: String, + obj: String, + localFilePath: String, + normalizePath: String => String): Unit = { + logger.info(s"begin to download bucket = ${bucket}, object = ${obj}, localFilePath = ${localFilePath}.") + GZIPUtilV2.streamCreateTarArchive(localFilePath) { + tar => { + val f = { + (item: Item) => + val in = minioClient.getObject(bucket, item.objectName()) + try { + GZIPUtilV2.streamAddToArchive(in, item.objectSize(), normalizePath(item.objectName()), tar) + } finally { + in.close() + } + } + visit(bucket, obj, f) + } + } + } + +// def resovleUri(uri: URI): (String, String) = { +// val path = uri.getPath.split("/") +// (path.head, path.slice(1, path.length).mkString("/")) +// } + + /** + * visit given path, bucket/directory. if the bucket doesn't exist, than throw IllegalArgumentException. + * @param bucket the bucket to visit. + * @param obj the object in the bucket to visit. + */ + def visit(bucket: String, + obj: String, + f: Item => Unit): Unit = { + if (minioClient.bucketExists(bucket)) { + minioClient.listObjects(bucket, obj) + .asScala + .map(result => { + val item = result.get() + if (item.isDir) { + visit(bucket, item.objectName(), f) + } else { + f(item) + } + }) + } else { + throw new IllegalArgumentException(s"unknown bucket ${bucket}.") + } + + } + +} + +case class S3ServerInfo( + endpoint: String, + accessKeyId: String, + secretAccessKey: String) + +object MLFlowMinioUtilImpl extends MinioUtil { + override def serverInfo: S3ServerInfo = S3ServerInfo( + sys.env.getOrElse("MLFLOW_S3_ENDPOINT_URL", throw new RuntimeException()), + sys.env.getOrElse("AWS_ACCESS_KEY_ID", throw new RuntimeException()), + sys.env.getOrElse("AWS_SECRET_ACCESS_KEY", throw new RuntimeException()) + ) + + // normalize the sub object path in the root obj. drop the experiment id in the path head. + // e.g. + // /0/a063487ee34e463baf7101d145b96bb7/artifacts => /a063487ee34e463baf7101d145b96bb7/artifacts + val _normalizePath = { + (absolutePathInBucket: String) => + absolutePathInBucket.split("/") match { + case Array(_, _, normalized @ _*) => normalized.mkString("/", "/", "") + } + } + + override def downloadAsZip(bucket: String, + obj: String, + localFilePath: String, + normalizePath: String => String = _normalizePath): Unit = { + super.downloadAsZip(bucket, obj, localFilePath, normalizePath) + } +} + +object MinioUtilTest extends MinioUtil { + override def serverInfo: S3ServerInfo = S3ServerInfo( + "http://192.168.218.59:9000", + "AKIAIOSFODNN7EXAMPLE", + "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY" + ) + + def main(args: Array[String]): Unit = { + val a = "s3://test/0/a063487ee34e463baf7101d145b96bb7/artifacts" + val uri = URI.create(a) + } +} diff --git a/common/src/main/scala/org/apache/spark/panda/utils/SFTPUtil.scala b/common/src/main/scala/org/apache/spark/panda/utils/SFTPUtil.scala index a867ccd..e6a6c06 100644 --- a/common/src/main/scala/org/apache/spark/panda/utils/SFTPUtil.scala +++ b/common/src/main/scala/org/apache/spark/panda/utils/SFTPUtil.scala @@ -36,7 +36,13 @@ object SFTPUtil { val config = configs.getConfig(host) val key = ssh.loadKeys(config.getValue("IdentityFile").replaceFirst("~", userHomePath)) val hostname = config.getHostname - val port = config.getPort + val port = { + if (config.getPort != -1) { + config.getPort + } else { + 22 + } + } try { ssh.connect(hostname, port) ssh.authPublickey(config.getUser, key) diff --git a/common/src/main/scala/org/apache/spark/panda/utils/Util.scala b/common/src/main/scala/org/apache/spark/panda/utils/Util.scala index 5051953..fd2781b 100644 --- a/common/src/main/scala/org/apache/spark/panda/utils/Util.scala +++ b/common/src/main/scala/org/apache/spark/panda/utils/Util.scala @@ -1,7 +1,7 @@ package org.apache.spark.panda.utils -import java.io.File -import java.nio.file.{Path, Paths} +import java.io.{File, IOException} +import java.nio.file.{Files, Path, Paths} import java.security.MessageDigest /** @@ -51,6 +51,15 @@ object Util { these ++ these.filter(_.isDirectory).flatMap(recursiveListFiles) } + @throws(classOf[IOException]) + def recursiveDeleteFile(path: String): Unit = { + recursiveListFiles(Paths.get(path).toFile) + .foreach(f => { + Files.deleteIfExists(f.toPath) + }) + Files.deleteIfExists(Paths.get(path)) + } + def stringToMD5(string: String): String = { MessageDigest.getInstance("MD5") .digest(string.getBytes("UTF-8")) @@ -58,5 +67,12 @@ object Util { .mkString } + def mkdir(path: String): Unit = { + val p = Paths.get(path) + if (!Files.exists(p)) { + Files.createDirectories(p) + } + } + } diff --git a/common/src/main/scala/org/panda/Config.scala b/common/src/main/scala/org/panda/Config.scala new file mode 100644 index 0000000..233a037 --- /dev/null +++ b/common/src/main/scala/org/panda/Config.scala @@ -0,0 +1,9 @@ +package org.panda + +/** + * @time 2020/3/9 4:42 下午 + * @author fchen + */ +object Config { + val CACHE_ROOT_DIR = sys.env.getOrElse("BAMBOO_CACHE_DIR", "/tmp/cache") +} diff --git a/common/src/test/scala/org/apache/spark/panda/utils/CondaSuite.scala b/common/src/test/scala/org/apache/spark/panda/utils/CondaSuite.scala new file mode 100644 index 0000000..e377214 --- /dev/null +++ b/common/src/test/scala/org/apache/spark/panda/utils/CondaSuite.scala @@ -0,0 +1,58 @@ +package org.apache.spark.panda.utils + +import org.scalatest.FunSuite + +/** + * @time 2020/1/3 下午1:52 + * @author fchen + */ +class CondaSuite extends FunSuite { + test("basic - the same libary in dependencies and pip should return different environment name.") { + val yaml1 = + """ + |channels: + |- defaults + |dependencies: + |- python=3.7.4 + |- lightgbm==2.2.3 + |- pip: + | - mlflow + | - cloudpickle==1.2.2 + |name: mlflow-env + |""".stripMargin + + val yaml2 = + """ + |channels: + |- defaults + |dependencies: + |- python=3.7.4 + |- pip: + | - mlflow + | - lightgbm==2.2.3 + | - cloudpickle==1.2.2 + |name: mlflow-env + |""".stripMargin + + val name1 = Conda.normalize(yaml1).get("name") + val name2 = Conda.normalize(yaml2).get("name") + assert(name1 != name2) + + // case 2 + val yaml3 = + """ + |dependencies: + |- python=3.7.4 + |""".stripMargin + val yaml4 = + """ + |dependencies: + |- pip: + | - python=3.7.4 + |""".stripMargin + val name3 = Conda.normalize(yaml3, false).get("name") + val name4 = Conda.normalize(yaml4, false).get("name") + assert(name3 != name4) + } + +} diff --git a/core/pom.xml b/core/pom.xml index 1316d35..000dd03 100644 --- a/core/pom.xml +++ b/core/pom.xml @@ -30,6 +30,10 @@ org.apache.spark spark-sql_${scala.binary.version} + + org.mlflow + mlflow-client + diff --git a/core/src/main/python/dump_pyfunc.py b/core/src/main/python/dump_pyfunc.py index 2a0327e..43f98c8 100644 --- a/core/src/main/python/dump_pyfunc.py +++ b/core/src/main/python/dump_pyfunc.py @@ -13,6 +13,20 @@ print("function return type: " + str(return_type)) archive_path = sys.argv[3] +class ModelCache(object): + _models = {} + + def __init__(self): + pass + + @staticmethod + def get_or_load(archive_path): + if archive_path in ModelCache._models: + return ModelCache._models[archive_path] + from mlflow.pyfunc import load_pyfunc + ModelCache._models[archive_path] = load_pyfunc(archive_path) + return ModelCache._models[archive_path] + def predict(*args): import pandas from mlflow.pyfunc.spark_model_cache import SparkModelCache @@ -30,11 +44,12 @@ def predict(*args): message="Invalid result_type '{}'. Result type can only be one of or an array of one " "of the following types types: {}".format(str(elem_type), str(supported_types)), error_code=INVALID_PARAMETER_VALUE) - model = SparkModelCache.get_or_load(archive_path) + # model = SparkModelCache.get_or_load(archive_path) # model = load_pyfunc(archive_path) - schema = {str(i): arg for i, arg in enumerate(args)} + model = ModelCache.get_or_load(archive_path) + schema = {series.name: series for i, series in enumerate(args)} # Explicitly pass order of columns to avoid lexicographic ordering (i.e., 10 < 2) - columns = [str(i) for i, _ in enumerate(args)] + columns = [series.name for i, series in enumerate(args)] pdf = pandas.DataFrame(schema, columns=columns) # model.predict(pdf) result = model.predict(pdf) diff --git a/core/src/main/scala/org/apache/spark/catalyst/parser/CreateFunctionParser.scala b/core/src/main/scala/org/apache/spark/catalyst/parser/CreateFunctionParser.scala index 43ff3c3..c0a577d 100644 --- a/core/src/main/scala/org/apache/spark/catalyst/parser/CreateFunctionParser.scala +++ b/core/src/main/scala/org/apache/spark/catalyst/parser/CreateFunctionParser.scala @@ -6,47 +6,49 @@ import scala.collection.JavaConverters._ import org.apache.spark.catalyst.parser.CreateFunctionParser.ExtensionsBuilder import org.apache.spark.sql.{SparkSession, SparkSessionExtensions} -import org.apache.spark.sql.catalyst.parser.{AbstractSqlParser, AstBuilder, ParseException, ParserInterface} +import org.apache.spark.sql.catalyst.parser.{AbstractSqlParser, AstBuilder, ParseException, ParserInterface, SqlBaseParser} import org.apache.spark.sql.catalyst.parser.ParserUtils._ import org.apache.spark.sql.catalyst.parser.SqlBaseParser.{CreateFunctionContext, QualifiedNameContext} import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan -import org.apache.spark.sql.catalyst.trees.Origin +import org.apache.spark.sql.catalyst.rules.ChangePythonUDFStrategy +import org.apache.spark.sql.execution.SparkSqlAstBuilder import org.apache.spark.sql.execution.command.{CreateFunctionCommand, CreateMLFlowFunctionCommand} -import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.execution.python.OneRowRelationToLocalRelation +import org.apache.spark.sql.internal.{SQLConf, VariableSubstitution} /** * @time 2019-09-05 14:00 * @author fchen */ -class CreateFunctionParser() extends AbstractSqlParser { +class CreateFunctionParser(conf: SQLConf) extends AbstractSqlParser { - override def parsePlan(sqlText: String): LogicalPlan = { - parse(sqlText) { parser => - astBuilder.visitSingleStatement(parser.singleStatement()) match { - case plan: LogicalPlan => plan - case _ => - val position = Origin(None, None) - throw new ParseException(Option(sqlText), "Unsupported SQL statement", position, position) - } - } + protected override def parse[T](command: String)(toResult: SqlBaseParser => T): T = { + super.parse(substitutor.substitute(command))(toResult) } - override protected def astBuilder: AstBuilder = new PandaAstBuider(new SQLConf()) + + private val substitutor = new VariableSubstitution(conf) + + override protected def astBuilder: AstBuilder = new PandaAstBuider(conf) } object CreateFunctionParser { type ParserBuilder = (SparkSession, ParserInterface) => ParserInterface type ExtensionsBuilder = SparkSessionExtensions => Unit - val parserBuilder: ParserBuilder = (_, _) => new CreateFunctionParser() + val parserBuilder: ParserBuilder = (_, _) => new CreateFunctionParser(new SQLConf) val extBuilder: ExtensionsBuilder = { e => e.injectParser(parserBuilder)} } class PandaSparkExtensions extends ExtensionsBuilder { override def apply(sessionExtensions: SparkSessionExtensions): Unit = { - sessionExtensions.injectParser((_, _) => new CreateFunctionParser()) + sessionExtensions.injectParser((_, _) => new CreateFunctionParser(new SQLConf)) + sessionExtensions.injectPlannerStrategy(_ => new ChangePythonUDFStrategy) +// sessionExtensions.injectOptimizerRule(_ => new OneRowRelationToLocalRelation) +// sessionExtensions.injectPostHocResolutionRule(_ => new OneRowRelationToLocalRelation) + sessionExtensions.injectResolutionRule(_ => new OneRowRelationToLocalRelation) } } -class PandaAstBuider(conf: SQLConf) extends AstBuilder(conf) { +class PandaAstBuider(conf: SQLConf) extends SparkSqlAstBuilder(conf) { override def visitCreateFunction(ctx: CreateFunctionContext): LogicalPlan = withOrigin(ctx) { val options = ctx.resource.asScala.map { resource => @@ -67,7 +69,7 @@ class PandaAstBuider(conf: SQLConf) extends AstBuilder(conf) { ctx.REPLACE != null) } else { - super.visitCreateFunction(ctx).asInstanceOf[LogicalPlan] + super.visitCreateFunction(ctx) } } } diff --git a/core/src/main/scala/org/apache/spark/sql/catalyst/rules/ChangePythonUDFRule.scala b/core/src/main/scala/org/apache/spark/sql/catalyst/rules/ChangePythonUDFRule.scala new file mode 100644 index 0000000..5cc1316 --- /dev/null +++ b/core/src/main/scala/org/apache/spark/sql/catalyst/rules/ChangePythonUDFRule.scala @@ -0,0 +1,73 @@ +package org.apache.spark.sql.catalyst.rules + +import org.apache.spark.sql.{SparkSession, Strategy} +import org.apache.spark.sql.catalyst.expressions.{Expression, NamedExpression, PythonUDF} +import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Project} +import org.apache.spark.sql.execution.{ProjectExec, RDDScanExec, SparkPlan} +import org.apache.spark.sql.execution.python.{ArrowEvalPython, KFArrowEvalPython, KFArrowEvalPythonExec} +import org.apache.spark.sql.types.{DataTypes, StructField, StructType} + +/** + * @time 2020/1/9 1:32 下午 + * @author fchen + */ + +class ChangePythonUDFStrategy extends Strategy { + override def apply(plan: LogicalPlan): Seq[SparkPlan] = { + plan match { + case ArrowEvalPython(udfs, output, child) => + val inputSchema = { + val is = findInputSchema2(udfs) + if (is.size == 0) None else Option(StructType(is.distinct)) + } + val s = findInputSchema(udfs) + KFArrowEvalPythonExec(udfs, inputSchema, output, planLater(child)) :: Nil + case _ => + Nil + } + } + + def findInputSchema(expressions: Seq[Expression]): Option[StructType] = { + if (expressions == null || expressions.size == 0) { + None + } else if (expressions.head.children.forall(_.isInstanceOf[PythonUDF])) { + findInputSchema(expressions.head.children) + } else { + if (expressions.head.children.forall(_.isInstanceOf[NamedExpression])) { + Option(StructType( + expressions.head.children.map(_.asInstanceOf[NamedExpression]).map(ne => { + StructField(ne.name, ne.dataType) + }) + )) + } else { + None + } + } + + Option(StructType( + Seq(StructField("x", DataTypes.IntegerType), + StructField("y", DataTypes.IntegerType) + ) + )) + } + + def findInputSchema2(expressions: Seq[Expression]): Seq[StructField] = { + if (expressions == null || expressions.size == 0) { + Seq.empty + } else { + expressions.flatMap(expression => { + if (expression.children.forall(_.isInstanceOf[PythonUDF])) { + findInputSchema2(expression.children) + } else { + if (expression.children.forall(_.isInstanceOf[NamedExpression])) { + expression.children.map(_.asInstanceOf[NamedExpression]).map(ne => { + StructField(ne.name, ne.dataType) + }) + } else { + Seq.empty[StructField] + } + } + }) + } + } +} diff --git a/core/src/main/scala/org/apache/spark/sql/execution/command/CreateMLFlowFunctionCommand.scala b/core/src/main/scala/org/apache/spark/sql/execution/command/CreateMLFlowFunctionCommand.scala index 1950dc5..a6280f7 100644 --- a/core/src/main/scala/org/apache/spark/sql/execution/command/CreateMLFlowFunctionCommand.scala +++ b/core/src/main/scala/org/apache/spark/sql/execution/command/CreateMLFlowFunctionCommand.scala @@ -1,6 +1,12 @@ package org.apache.spark.sql.execution.command +import java.util.Base64 + +import org.apache.spark.{SparkConf, SparkFiles} +import org.apache.spark.internal.config.ConfigBuilder +import org.apache.spark.panda.utils.{Conda, MLmodelParser} import org.apache.spark.sql.{Row, SparkSession} +import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.panda.PandasFunctionManager import org.apache.spark.sql.types.DataType @@ -18,17 +24,90 @@ case class CreateMLFlowFunctionCommand( replace: Boolean) extends RunnableCommand { override def run(sparkSession: SparkSession): Seq[Row] = { - val pythonExec = options.get("pythonexec") - val pythonVer = options.get("pythonver") - PandasFunctionManager.registerMLFlowPythonUDF( - sparkSession, functionName, - returnType = Option(DataType.fromDDL(options("returns"))), - artifactRoot = Option(options("artifactroot")), - runId = className, - driverPythonExec = pythonExec, - driverPythonVer = pythonVer, - pythonExec = pythonExec, - pythonVer = pythonVer) +// val pythonExec = options.get("pythonexec") +// val pythonVer = options.get("pythonver") +// PandasFunctionManager.registerMLFlowPythonUDF( +// sparkSession, functionName, +// returnType = Option(DataType.fromDDL(options("returns"))), +// artifactRoot = Option(options("artifactroot")), +// runId = className, +// driverPythonExec = pythonExec, +// driverPythonVer = pythonVer, +// pythonExec = pythonExec, +// pythonVer = pythonVer) + setup(sparkSession) + Seq.empty[Row] + } + + def setup(sparkSession: SparkSession): Unit = { + + // todo: 下载通过bamboo + + if (CreateMLFlowFunctionCommand.isBambooServerEnable) { + val runid = options.getOrElse("runid", className) + // first we download mlflow run from bamboo server and parser MLmodel file. + val run = s"http://${CreateMLFlowFunctionCommand.bambooServer}/api/v1/artifact/createAndGet/${runid}/${runid}.tgz" + sparkSession.sparkContext.addFile(run) + val mlmodelPath = SparkFiles.get(runid) + s"/artifacts/model/MLmodel" + val content = scala.io.Source.fromFile(mlmodelPath) + .getLines() + .mkString("\n") + val mlmodel = new MLmodelParser(content) + + // second we download the python environment from bamboo server with the conda configurations. + val condaConfPath = SparkFiles.get(runid) + s"/artifacts/${mlmodel.artifactPath}/${mlmodel.env}" + val condaYaml = scala.io.Source.fromFile(condaConfPath) + .getLines() + .mkString("\n") + val name = Conda.normalize(condaYaml).get("name").toString + val encodeConf = Base64.getEncoder.encodeToString(condaYaml.getBytes("utf-8")) + val condaUrl = + s"http://${CreateMLFlowFunctionCommand.bambooServer}/api/v1/conda/createAndGet/${encodeConf}/${name}.tgz" + sparkSession.sparkContext.addFile(condaUrl) + + val driverPython = s"${SparkFiles.get(name)}/bin/python" + val pythonPath = s"./${name}/bin/python" + val pythonExec = Option(pythonPath) + val pythonVer = options.get("pythonver") + + PandasFunctionManager.registerMLFlowPythonUDFLocal( + sparkSession, + functionName, + s"./${runid}/artifacts/${mlmodel.artifactPath}", + returnType = Option(DataType.fromDDL(options("returns"))), + driverPythonExec = Option(driverPython), + driverPythonVer = pythonVer, + pythonExec = pythonExec, + pythonVer = pythonVer) + } else { + val pythonExec = options.get("pythonexec") + PandasFunctionManager.registerMLFlowPythonUDFLocal( + sparkSession, + functionName, + options.getOrElse("modellocalpath", ""), + returnType = Option(DataType.fromDDL(options("returns"))), + driverPythonExec = options.get("driverpythonexec").orElse(pythonExec), + driverPythonVer = options.get("pythonver"), + pythonExec = pythonExec, + pythonVer = options.get("pythonver")) + } Seq.empty[Row] } + +} + +object CreateMLFlowFunctionCommand { + val PANDA_BAMBOO_SERVER_ENABLE = SQLConf.buildConf("spark.panda.bamboo.server.enable") + .booleanConf + .createWithDefault(true) + + val PANDA_BAMBOO_SERVER = SQLConf.buildConf("spark.panda.bamboo.server") + .stringConf + .checkValue(address => !(isBambooServerEnable && address == ""), + "can't find spark.panda.bamboo.server in spark conf, " + + "please make sure you have set right configurations" + ).createWithDefaultString("") + val bambooServer = SQLConf.get.getConf(PANDA_BAMBOO_SERVER) + + val isBambooServerEnable = SQLConf.get.getConf(PANDA_BAMBOO_SERVER_ENABLE) } diff --git a/core/src/main/scala/org/apache/spark/sql/execution/python/KFEvalPythonExec.scala b/core/src/main/scala/org/apache/spark/sql/execution/python/KFEvalPythonExec.scala new file mode 100644 index 0000000..7256874 --- /dev/null +++ b/core/src/main/scala/org/apache/spark/sql/execution/python/KFEvalPythonExec.scala @@ -0,0 +1,187 @@ +package org.apache.spark.sql.execution.python + +import java.io.File + +import scala.collection.JavaConverters._ +import scala.collection.mutable.ArrayBuffer + +import org.apache.spark.{SparkEnv, TaskContext} +import org.apache.spark.api.python.{ChainedPythonFunctions, PythonEvalType} +import org.apache.spark.rdd.RDD +import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.catalyst.expressions._ +import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Filter, LeafNode, LocalRelation, LogicalPlan, OneRowRelation, Project, Statistics, SubqueryAlias, UnaryNode} +import org.apache.spark.sql.catalyst.rules.Rule +import org.apache.spark.sql.execution.{LeafExecNode, SparkPlan} +import org.apache.spark.sql.execution.arrow.ArrowUtils +import org.apache.spark.sql.types.{DataType, StructField, StructType} +import org.apache.spark.util.Utils + +/** + * @time 2020/1/9 1:18 下午 + * @author fchen + * + * copy from spark. + */ +abstract class KFEvalPythonExec(udfs: Seq[PythonUDF], + inputSchema: Option[StructType], + output: Seq[Attribute], + child: SparkPlan) extends EvalPythonExec(udfs, output, child) { + protected override def doExecute(): RDD[InternalRow] = { + val inputRDD = child.execute().map(_.copy()) + + inputRDD.mapPartitions { iter => + val context = TaskContext.get() + + // The queue used to buffer input rows so we can drain it to + // combine input with output from Python. + val queue = HybridRowQueue(context.taskMemoryManager(), + new File(Utils.getLocalDir(SparkEnv.get.conf)), child.output.length) + context.addTaskCompletionListener[Unit] { ctx => + queue.close() + } + + val (pyFuncs, inputs) = udfs.map(collectFunctions).unzip + + // flatten all the arguments + val allInputs = new ArrayBuffer[Expression] + val dataTypes = new ArrayBuffer[DataType] + val argOffsets = inputs.map { input => + input.map { e => + if (allInputs.exists(_.semanticEquals(e))) { + allInputs.indexWhere(_.semanticEquals(e)) + } else { + allInputs += e + dataTypes += e.dataType + allInputs.length - 1 + } + }.toArray + }.toArray + val projection = newMutableProjection(allInputs, child.output) + +// val schema = if (child.schema == null) { +// StructType(dataTypes.zipWithIndex.map { case (dt, i) => +// StructField(s"_$i", dt) +// }) +// } else { +// child.schema +// } + + val schema = inputSchema.getOrElse { + StructType(dataTypes.zipWithIndex.map { case (dt, i) => + StructField(s"_$i", dt) + }) + } + // 为什么这里能拿到正确的output？ + schema.printTreeString() + + // Add rows to queue to join later with the result. + val projectedRowIter = iter.map { inputRow => + queue.add(inputRow.asInstanceOf[UnsafeRow]) + projection(inputRow) + } + + val outputRowIterator = evaluate( + pyFuncs, argOffsets, projectedRowIter, schema, context) + + val joined = new JoinedRow + val resultProj = UnsafeProjection.create(output, output) + + outputRowIterator.map { outputRow => + resultProj(joined(queue.remove(), outputRow)) + } + } + } + private def collectFunctions(udf: PythonUDF): (ChainedPythonFunctions, Seq[Expression]) = { + udf.children match { + case Seq(u: PythonUDF) => + val (chained, children) = collectFunctions(u) + (ChainedPythonFunctions(chained.funcs ++ Seq(udf.func)), children) + case children => + // There should not be any other UDFs, or the children can't be evaluated directly. + assert(children.forall(_.find(_.isInstanceOf[PythonUDF]).isEmpty)) + (ChainedPythonFunctions(Seq(udf.func)), udf.children) + } + } + +} + +/** + * A physical plan that evaluates a [[PythonUDF]]. + */ +case class KFArrowEvalPythonExec(udfs: Seq[PythonUDF], + inputSchema: Option[StructType], + output: Seq[Attribute], + child: SparkPlan) + extends KFEvalPythonExec(udfs, inputSchema, output, child) { + + private val batchSize = conf.arrowMaxRecordsPerBatch + private val sessionLocalTimeZone = conf.sessionLocalTimeZone + private val pythonRunnerConf = ArrowUtils.getPythonRunnerConfMap(conf) + + protected override def evaluate(funcs: Seq[ChainedPythonFunctions], + argOffsets: Array[Array[Int]], + iter: Iterator[InternalRow], + schema: StructType, + context: TaskContext): Iterator[InternalRow] = { + + val outputTypes = output.drop(child.output.length).map(_.dataType) + + // DO NOT use iter.grouped(). See BatchIterator. + val batchIter = if (batchSize > 0) new BatchIterator(iter, batchSize) else Iterator(iter) + + val columnarBatchIter = new ArrowPythonRunner( + funcs, + PythonEvalType.SQL_SCALAR_PANDAS_UDF, + argOffsets, + schema, + sessionLocalTimeZone, + pythonRunnerConf).compute(batchIter, context.partitionId(), context) + + new Iterator[InternalRow] { + + private var currentIter = if (columnarBatchIter.hasNext) { + val batch = columnarBatchIter.next() + val actualDataTypes = (0 until batch.numCols()).map(i => batch.column(i).dataType()) + assert(outputTypes == actualDataTypes, "Invalid schema from pandas_udf: " + + s"expected ${outputTypes.mkString(", ")}, got ${actualDataTypes.mkString(", ")}") + batch.rowIterator.asScala + } else { + Iterator.empty + } + + override def hasNext: Boolean = currentIter.hasNext || { + if (columnarBatchIter.hasNext) { + currentIter = columnarBatchIter.next().rowIterator.asScala + hasNext + } else { + false + } + } + + override def next(): InternalRow = currentIter.next() + } + } +} + +case class OneRowRelationToLocalRelation() extends Rule[LogicalPlan] { + override def apply(plan: LogicalPlan): LogicalPlan = { + plan transform { + case proj@ Project(projectList, relation: OneRowRelation) if projectList.forall(_.resolved) => + val output = projectList.map(_.toAttribute) + val data = projectList.map { + case as: Alias => + as.child.eval() + } + Project(projectList, LocalRelation(output, Seq(InternalRow(data)))) + } + } +} + +/** + * A logical plan that evaluates a [[PythonUDF]]. + */ +case class KFArrowEvalPython(udfs: Seq[PythonUDF], + inputSchema: StructType, + output: Seq[Attribute], child: LogicalPlan) + extends UnaryNode diff --git a/core/src/main/scala/org/apache/spark/sql/panda/PandasFunctionManager.scala b/core/src/main/scala/org/apache/spark/sql/panda/PandasFunctionManager.scala index 5e3acd8..d86a4e2 100644 --- a/core/src/main/scala/org/apache/spark/sql/panda/PandasFunctionManager.scala +++ b/core/src/main/scala/org/apache/spark/sql/panda/PandasFunctionManager.scala @@ -38,18 +38,17 @@ object PandasFunctionManager { } def registerMLFlowPythonUDFLocal(spark: SparkSession, - functionName: String, - modelLocalPath: String, - returnType: Option[DataType] = None, - driverPythonExec: Option[String] = None, - driverPythonVer: Option[String] = None, - pythonExec: Option[String] = None, - pythonVer: Option[String] = None - ): Unit = { - val modelPath = SparkModelCache.addLocalModel(spark, modelLocalPath) + functionName: String, + modelLocalPath: String, + returnType: Option[DataType] = None, + driverPythonExec: Option[String] = None, + driverPythonVer: Option[String] = None, + pythonExec: Option[String] = None, + pythonVer: Option[String] = None): Unit = { +// val modelPath = SparkModelCache.addLocalModel(spark, modelLocalPath) val funcSerPath = Utils.createTempDir().getPath + File.separator + "dump_func" writeBinaryPythonFunc( - funcSerPath, modelPath, returnType.getOrElse(IntegerType), + funcSerPath, modelLocalPath, returnType.getOrElse(IntegerType), driverPythonExec.getOrElse("python") ) registerPythonUDF(spark, funcSerPath, functionName, returnType, pythonExec, pythonVer) diff --git a/dev/bamboo/Dockerfile b/dev/bamboo/Dockerfile new file mode 100644 index 0000000..e87ef18 --- /dev/null +++ b/dev/bamboo/Dockerfile @@ -0,0 +1,36 @@ +FROM registry.uc.host.dxy/library/maven-conda + +# Define working directory. +WORKDIR /work + +# Prepare download dependencies +ADD . /work/ + +# setup conda and pip proxy. +RUN echo 'default_channels:\n\ + - http://nexus.k8s.uc.host.dxy/repository/anaconda/pkgs/msys2/\n\ + - http://nexus.k8s.uc.host.dxy/repository/anaconda/pkgs/pro/\n\ + - http://nexus.k8s.uc.host.dxy/repository/anaconda/pkgs/r/\n\ + - http://nexus.k8s.uc.host.dxy/repository/anaconda/pkgs/free/\n\ + - http://nexus.k8s.uc.host.dxy/repository/anaconda/pkgs/main/' > ~/.condarc + +RUN mkdir ~/.pip +RUN echo '[global]\n\ +index-url = http://nexus.k8s.uc.host.dxy/repository/pypi-aliyun/simple\n\ +trusted-host=nexus.k8s.uc.host.dxy' > ~/.pip/pip.conf + +ENV LANG C.UTF-8 +ENV MALLOC_ARENA_MAX 4 +ENV CONDA_PATH /opt/conda/bin/conda +ENV MLFLOW_TRACKING_URI http://mlflow.k8s.uc.host.dxy/ + +RUN apt-get update && apt-get install -y git + +# package +RUN mvn -gs settings.xml clean package -DskipTests -Passembly -Dassembly.target.module=kungfu-panda-bamboo -Dassembly.format=dir + +# clear maven cache. +RUN rm -rf ~/.m2 + +# Define default command. +ENTRYPOINT bash kungfu-panda-bin-1.0/kungfu-panda/bootstrap.sh run -Xmx4g -XX:+UseG1GC -Dserver.port=8100 org.panda.bamboo.service.Application diff --git a/dev/maven-conda/Dockerfile b/dev/maven-conda/Dockerfile new file mode 100644 index 0000000..b3264a8 --- /dev/null +++ b/dev/maven-conda/Dockerfile @@ -0,0 +1,10 @@ +FROM maven:3.6.3-jdk-8 + +# install miniconda +RUN wget --quiet https://repo.anaconda.com/miniconda/Miniconda3-4.5.11-Linux-x86_64.sh -O ~/miniconda.sh && \ + /bin/bash ~/miniconda.sh -b -p /opt/conda && \ + rm ~/miniconda.sh && \ + /opt/conda/bin/conda clean -tipsy && \ + ln -s /opt/conda/etc/profile.d/conda.sh /etc/profile.d/conda.sh && \ + echo ". /opt/conda/etc/profile.d/conda.sh" >> ~/.bashrc && \ + echo "conda activate base" >> ~/.bashrc diff --git a/examples/local/pom.xml b/examples/local/pom.xml index 538f217..7f73e2b 100644 --- a/examples/local/pom.xml +++ b/examples/local/pom.xml @@ -34,6 +34,11 @@ org.apache.spark spark-mllib_${scala.binary.version} + + + + + diff --git a/examples/local/src/main/scala/org/panda/example/local/PandaSqlExample.scala b/examples/local/src/main/scala/org/panda/example/local/PandaSqlExample.scala index 7daad04..e3e1865 100644 --- a/examples/local/src/main/scala/org/panda/example/local/PandaSqlExample.scala +++ b/examples/local/src/main/scala/org/panda/example/local/PandaSqlExample.scala @@ -1,3 +1,4 @@ +// scalastyle:off package org.panda.example.local import java.io.{File, FileInputStream} @@ -6,85 +7,173 @@ import java.util.Base64 import com.google.common.io.ByteStreams import org.apache.spark.catalyst.parser.CreateFunctionParser import org.apache.spark.sql.{SparkSession, SparkSessionExtensions} +import org.apache.spark.sql.panda.PandasFunctionManager +import org.apache.spark.sql.types.StringType /** * @time 2019-09-05 14:59 * @author fchen */ -// scalastyle:off object PandaSqlExample { def main(args: Array[String]): Unit = { // - val path = "/tmp/testdata/12891819633_e4c82b51e8.jpg" - val file = new File(path) - val in = new FileInputStream(file) - val array = ByteStreams.toByteArray(in) - println(array.slice(0, 100).mkString(",")) -// println(new String(Base64.getEncoder.encode(array), "utf-8")) -// println(new String( org.apache.commons.codec.binary.Base64.encodeBase64( -// array -// ))) -//// println(array.slice(0, 100).mkString(",")) -// System.exit(0) - - + test() + System.exit(0) val spark = SparkSession .builder() .appName("panda sql example") .master("local[4]") .config("spark.sql.extensions", "org.apache.spark.catalyst.parser.PandaSparkExtensions") - + .config("spark.panda.bamboo.server.enable", "false") // .withExtensions(CreateFunctionParser.extBuilder) .getOrCreate() + val python = "/usr/local/share/anaconda3/envs/mlflow-study/bin/python" +// val python = "/usr/local/share/anaconda3/envs/pyspark-2.4.3/bin/python" + val path = "/Users/fchen/Project/fchen/kungfu-panda/examples/python/sklearn_kmeans/mlruns/1/e60af958648a4f7981c1195f82d82c1d/artifacts/model" + spark.sql( + s""" + |CREATE FUNCTION `test` AS '909e8c3a8b504f11ac29150af83cee42' USING + | `type` 'mlflow', + | `modelLocalPath` '$path', + | `pythonExec` '$python', + | `returns` 'int' + |""".stripMargin) import spark.implicits._ - import org.apache.spark.sql.functions._ - val binaryFile = spark.sparkContext.binaryFiles("/tmp/testdata/*") - .map(r => (r._1, r._2.toArray())) - .toDF("filename", "image") + Seq( + (11, 22), + (33, 44) + ).toDF("x", "y") + .repartition(1) + .selectExpr("test(x, y)") + .show() +// .explain(true) -// val df = spark.read.format("binaryFile") - val df = binaryFile - .selectExpr("cast(base64(image) as string) as feature", "filename") -// val df = spark.read -// .format("image") -// .load("/tmp/flower_photos/*") -//// .select($"value", input_file_name as "name") -//// .printSchema() -// .select("image.*") -// .selectExpr("base64(data) as feature_array", "origin", "data") -// .selectExpr("cast(feature_array as string) as feature", "xx(feature_array) as f2", "feature_array", "data", "yy(data)") -// df.show() -// System.exit(0) +// spark.sql( +// """ +// |select test(x, y) from ( +// |select 1 as x, 1 as y +// |) +// |""".stripMargin) // .show() -// val python = "/usr/local/share/anaconda3/envs/tensorflow-example/bin/python" - val python = "/usr/local/share/anaconda3/envs/flower_classifier/bin/python" - val artifactRoot = "/tmp" - val runid = "c1f48fc796f3467cb104114f3fa501df" -// val python = "/usr/local/share/anaconda3/envs/mlflow-study/bin/python" -// val artifactRoot = "/Users/fchen/Project/python/mlflow-study/mlruns" -// val runid = "9c6c59d0f57f40dfbbded01816896687" -// + } + + def test(): Unit = { + val spark = SparkSession + .builder() + .appName("panda sql example") + .master("local[4]") + .config("spark.sql.extensions", "org.apache.spark.catalyst.parser.PandaSparkExtensions") + .config("spark.sql.codegen.wholeStage", "false") + .config("spark.sql.execution.arrow.enabled", "true") + .config("spark.panda.bamboo.server.enable", "false") + // .withExtensions(CreateFunctionParser.extBuilder) + .getOrCreate() + val path = "/Users/fchen/Project/fchen/examples/mlflow-in-action/add/mlruns/1/58d234e03699404c938e0ba87d627920/artifacts/model" + val python = "/usr/local/share/anaconda3/envs/mlflow-study/bin/python" + // val python = "/usr/local/share/anaconda3/envs/pyspark-2.4.3/bin/python" +// val path = "/Users/fchen/Project/fchen/kungfu-panda/examples/python/sklearn_kmeans/mlruns/1/e60af958648a4f7981c1195f82d82c1d/artifacts/model" + spark.sql( + s""" + |CREATE FUNCTION `test` AS '909e8c3a8b504f11ac29150af83cee42' USING + | `type` 'mlflow', + | `modelLocalPath` '$path', + | `pythonExec` '$python', + | `returns` 'int' + |""".stripMargin) + + val path2 = "/Users/fchen/Project/fchen/examples/mlflow-in-action/add/mlruns/1/19131276fd084da5b0b629d62448f206/artifacts/model" + // val python = "/usr/local/share/anaconda3/envs/pyspark-2.4.3/bin/python" + // val path = "/Users/fchen/Project/fchen/kungfu-panda/examples/python/sklearn_kmeans/mlruns/1/e60af958648a4f7981c1195f82d82c1d/artifacts/model" spark.sql( s""" - |CREATE FUNCTION `test` AS '${runid}' USING - | `type` 'mlflow', - | `returns` 'array', - | `artifactRoot` '${artifactRoot}', - | `pythonExec` '${python}', - | `pythonVer` '3.7' - """.stripMargin) + |CREATE FUNCTION `test2` AS '909e8c3a8b504f11ac29150af83cee42' USING + | `type` 'mlflow', + | `modelLocalPath` '$path2', + | `pythonExec` '$python', + | `returns` 'int' + |""".stripMargin) - df.selectExpr("test(feature) as predict", "filename").show() +// val dd: Int => Int = (i: Int) => i + 11 +// spark.udf.register("dd", dd) // +// val ff: Int => Int = (i: Int) => i + 13 +// spark.udf.register("ff", ff) + +// spark.sql( +// """ +// |select ff(y) from ( +// |select dd(x) as y from ( +// |select 1223 as x, 13334 +// |)) +// |""".stripMargin) +// .explain(true) + +// val df = spark.sql( +// """ +// |select current_date() +// |""".stripMargin) + + val df = spark.sql( + """ + |select test(x) from ( + |select 5 as x + |) + |""".stripMargin + ) + df.explain(true) + df.show() + +// val df = spark.sql( +// """ +// |select test2(y) from ( +// |select test(x) as y from ( +// |select 1223 as x, 13334 +// |)) +// |""".stripMargin) +// df.explain(true) +// df.show() + println("------------") +// spark.sql("select 1").explain(true) + import spark.implicits._ +// val df = Seq( +// (11, 22), +// (33, 44) +// ).toDF("x", "y") +// .repartition(1) +// .selectExpr("*", "test(x)", "test2(y)", "test(x)") +// df.explain(true) +// df.show +// df.show // spark.sql( // """ -// |select test(x, y) from ( -// |select 1 as x, 1 as y +// |select x + 1 from ( +// |select 1 as x // |) // |""".stripMargin) +// .explain(true) // .show() + } + def badcase: Unit = { + // todo:(fchen) 嵌套下为什么会有问题 + val sql = + """ + |select test(test(x)) from ( + |select 1223 as x, 13334 as y + |) + |""".stripMargin + + val sql2 = + """ + |select *,x from ( + |select 1223 as x, 13334 + |) + |""".stripMargin + val sql3 = + """ + |select test(5) + |""".stripMargin } } diff --git a/examples/python/sklearn_kmeans/README.md b/examples/python/sklearn_kmeans/README.md new file mode 100644 index 0000000..3a0b369 --- /dev/null +++ b/examples/python/sklearn_kmeans/README.md @@ -0,0 +1,30 @@ +# Sklearn KMeans + +1.环境安装 +2.模型训练 +```bash +export MLFLOW_TRACKING_URI="http://192.168.218.172:9999" && /usr/local/share/anaconda3/envs/mlflow-sklearn2/bin/python sklearn_kmeans/new_train.py +``` +3.部署成REST API + +```bash +source activate mlflow-sklearn2 +export MLFLOW_TRACKING_URI="http://192.168.218.172:9999" && mlflow models serve -m runs:/a4dc870e46274ff28fce1a537abd07a0/model --no-conda +``` + +预测 +```bash +curl -X POST \ + http://localhost:5000/invocations \ + -H 'cache-control: no-cache' \ + -H 'content-type: application/json' \ + -H 'postman-token: d26ffff1-3cfd-fab7-114a-8976764b4985' \ + -d '{ + "data": [ + {"x": 1, + "y": 1} + ] +}' +``` + +4. Spark SQL分布式批量预测 diff --git a/examples/python/sklearn_kmeans/conda.yaml b/examples/python/sklearn_kmeans/conda.yaml new file mode 100644 index 0000000..cd9d529 --- /dev/null +++ b/examples/python/sklearn_kmeans/conda.yaml @@ -0,0 +1,18 @@ +name: mlflow-study +channels: + - http://nexus.k8s.uc.host.dxy/repository/anaconda/pkgs/main/ + - http://nexus.k8s.uc.host.dxy/repository/anaconda/pkgs/free/ + - http://nexus.k8s.uc.host.dxy/repository/anaconda/pkgs/r/ + - http://nexus.k8s.uc.host.dxy/repository/anaconda/pkgs/pro/ + - http://nexus.k8s.uc.host.dxy/repository/anaconda/pkgs/msys2/ +dependencies: + - python=3.6 + - scikit-learn=0.19.1 + - cloudpickle=0.6.1 + - numpy=1.14.3 + - pandas=0.22.0 + - pyspark=2.4.3 + - matplotlib=3.1.1 + - pip: + - mlflow + - pyarrow==0.12.1 diff --git a/examples/python/sklearn_kmeans/new_train.py b/examples/python/sklearn_kmeans/new_train.py new file mode 100644 index 0000000..fe8faa2 --- /dev/null +++ b/examples/python/sklearn_kmeans/new_train.py @@ -0,0 +1,34 @@ +from pandas import DataFrame + +Data = { + 'x': [25, 34, 22, 27, 33, 33, 31, 22, 35, 34, 67, 54, 57, 43, 50, 57, 59, 52, 65, 47, 49, 48, 35, 33, 44, 45, 38, + 43, 51, 46], + 'y': [79, 51, 53, 78, 59, 74, 73, 57, 69, 75, 51, 32, 40, 47, 53, 36, 35, 58, 59, 50, 25, 20, 14, 12, 20, 5, 29, 27, + 8, 7] + } + +df = DataFrame(Data, columns=['x', 'y']) +# print(df) +from sklearn.cluster import KMeans + +kmeans = KMeans(n_clusters=3).fit(df) + + +d2 = { + 'x': [100], + 'y': [100] +} +d = DataFrame(d2, columns=['x', 'y']) +print(kmeans.predict(d)) + +# import matplotlib.pyplot as plt +# df['cl'] = kmeans.labels_ +# df.plot.scatter('x', 'y', c='cl', colormap='gist_rainbow') +# +# plt.show() + +import mlflow +import mlflow.sklearn +mlflow.set_experiment("odep example") +with mlflow.start_run(): + mlflow.sklearn.log_model(kmeans, "model") diff --git a/examples/yarn/pom.xml b/examples/yarn/pom.xml index 7d39c1e..97afa5a 100644 --- a/examples/yarn/pom.xml +++ b/examples/yarn/pom.xml @@ -36,5 +36,18 @@ + + + single-jar + + + + org.apache.maven.plugins + maven-shade-plugin + + + + + diff --git a/examples/yarn/src/main/scala/org/panda/example/yarn/KmeansExample.scala b/examples/yarn/src/main/scala/org/panda/example/yarn/KmeansExample.scala new file mode 100644 index 0000000..36dbfb0 --- /dev/null +++ b/examples/yarn/src/main/scala/org/panda/example/yarn/KmeansExample.scala @@ -0,0 +1,55 @@ +package org.panda.example.yarn + +import java.io.File + +import org.apache.spark.sql.SparkSession + +/** + * @time 2019/12/10 下午3:52 + * @author fchen + */ +object KmeansExample { + def main(args: Array[String]): Unit = { + val spark = SparkSession + .builder() + .appName("panda sql example") + .master("local[4]") + .config("spark.sql.extensions", "org.apache.spark.catalyst.parser.PandaSparkExtensions") + .config("spark.files.fetchTimeout", "600s") + .config("spark.panda.bamboo.server", "192.168.202.205:8888") +// .config("spark.panda.bamboo.server", "192.168.200.69:8100") + .getOrCreate() + + import spark.implicits._ + + // val python = "/usr/local/share/anaconda3/envs/tensorflow-example/bin/python" + val runid = "a063487ee34e463baf7101d145b96bb7" + // val python = "/usr/local/share/anaconda3/envs/mlflow-study/bin/python" + // val artifactRoot = "/Users/fchen/Project/python/mlflow-study/mlruns" + // val runid = "9c6c59d0f57f40dfbbded01816896687" + // + spark.sql("set spark.sql.crossJoin.enabled = true") + + spark.sql( + s""" + |CREATE FUNCTION `test` AS '${runid}' USING + | `type` 'mlflow', + | `returns` 'int' + """.stripMargin) + + spark.sql( + s""" + |CREATE FUNCTION `test` AS '${runid}' USING + | `type` 'mlflow', + | `returns` 'int' + """.stripMargin) + + spark.sql( + """ + |select test(x, y) from ( + |select 1 as x, 1 as y + |) + |""".stripMargin) + .show() + } +} diff --git a/examples/yarn/src/main/scala/org/panda/example/yarn/Test.scala b/examples/yarn/src/main/scala/org/panda/example/yarn/Test.scala index d7038ca..9cd4301 100644 --- a/examples/yarn/src/main/scala/org/panda/example/yarn/Test.scala +++ b/examples/yarn/src/main/scala/org/panda/example/yarn/Test.scala @@ -35,16 +35,19 @@ // .config(conf) // .getOrCreate() // val python = "/usr/local/share/anaconda3/envs/mlflow-study/bin/python" -// val pythonExec = Option(python) -// PandasFunctionManager.registerMLFlowPythonUDF(spark, functionName = "test", "", -// returnType = Option(IntegerType), pythonExec = pythonExec) -// -// spark.sql( -// """ -// |select test(x, y) from ( -// |select 1 as x, 1 as y -// |) -// |""".stripMargin) -// .show() +// spark.sparkContext.addFile("http://192.168.218.12:8080/api/v1/test/kp") +//// val pythonExec = Option(python) +//// PandasFunctionManager.registerMLFlowPythonUDF(spark, functionName = "test", "", +//// returnType = Option(IntegerType), pythonExec = pythonExec) +//// +//// spark.sql( +//// """ +//// |select test(x, y) from ( +//// |select 1 as x, 1 as y +//// |) +//// |""".stripMargin) +//// .show() +// spark.sql("select 1").show() +// Thread.sleep(Int.MaxValue) // } //} diff --git a/pom.xml b/pom.xml index 6899188..393c7a1 100644 --- a/pom.xml +++ b/pom.xml @@ -21,6 +21,19 @@ + + + aliyun + Nexus Release Repository + http://maven.aliyun.com/nexus/content/groups/public + + true + + + true + + + 1.8 @@ -31,6 +44,7 @@ ${project.basedir}/libs 2.4.3 provided + 2.10.3 false @@ -41,6 +55,14 @@ compile + + spark-3.0 + + 3.0.0-preview2 + 2.12 + 2.12.10 + + single-jar @@ -81,6 +103,12 @@ + + assembly + + assembly + + @@ -89,26 +117,26 @@ org.apache.maven.plugins maven-jar-plugin 3.0.2 - - ${build.lib.path} - - - - maven-dependency-plugin - 3.0.1 - - - package - - copy-dependencies - - - provided - ${build.lib.path} - - - + + + + + + + + + + + + + + + + + + + org.apache.maven.plugins @@ -176,31 +204,31 @@ - - org.scalastyle - scalastyle-maven-plugin - 0.8.0 - - false - true - false - false - ${project.basedir}/src/main/scala - ${basedir}/src/test/scala - scalastyle-config.xml - ${basedir}/target/scalastyle-output.xml - ${project.build.sourceEncoding} - ${project.reporting.outputEncoding} - - - - package - - check - - - - + + + + + + + + + + + + + + + + + + + + + + + + + maven-clean-plugin 3.0.0 @@ -272,6 +300,26 @@ ${spark.version} ${spark.scope} + + org.mlflow + mlflow-client + 1.4.0 + + + io.minio + minio + 6.0.11 + + + com.fasterxml.jackson.module + jackson-module-scala_${scala.binary.version} + ${jackson.version} + + + com.fasterxml.jackson.core + jackson-databind + ${jackson.version} + @@ -279,7 +327,6 @@ org.scalatest scalatest_${scala.binary.version} 3.0.3 - test org.scala-lang @@ -293,6 +340,5 @@ bamboo examples/local examples/yarn - assembly diff --git a/settings.xml b/settings.xml new file mode 100755 index 0000000..c3bc7b7 --- /dev/null +++ b/settings.xml @@ -0,0 +1,13 @@ + + + + + dxy + * + dxy + http://nexus.k8s.uc.host.dxy/repository/maven-public/ + + +