@W-8256900@ read via local filesystem addendum to #521 (#522)

* file system

* wip

* fix

* wip

Author: gerashegalov

core/src/main/scala/com/salesforce/op/OpWorkflowModelReader.scala

@@ -33,20 +33,20 @@ package com.salesforce.op
 
 import java.io.File
 
-import com.salesforce.op.OpWorkflowModelReadWriteShared.{FieldNames => FN}
-import com.salesforce.op.OpWorkflowModelReadWriteShared.FieldNames._
 import com.salesforce.op.OpWorkflowModelReadWriteShared.DeprecatedFieldNames._
+import com.salesforce.op.OpWorkflowModelReadWriteShared.FieldNames._
+import com.salesforce.op.OpWorkflowModelReadWriteShared.{FieldNames => FN}
 import com.salesforce.op.features.{FeatureJsonHelper, OPFeature, TransientFeature}
 import com.salesforce.op.filters.{FeatureDistribution, RawFeatureFilterResults}
 import com.salesforce.op.stages.OpPipelineStageReaderWriter._
 import com.salesforce.op.stages._
-import org.zeroturnaround.zip.ZipUtil
 import org.apache.commons.io.IOUtils
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.{FileSystem, Path}
 import org.apache.hadoop.io.compress.CompressionCodecFactory
 import org.json4s.JsonAST.{JArray, JNothing, JValue}
 import org.json4s.jackson.JsonMethods.parse
+import org.zeroturnaround.zip.ZipUtil
 
 import scala.collection.mutable.ArrayBuffer
 import scala.io.Source
@@ -71,34 +71,39 @@ class OpWorkflowModelReader(val workflowOpt: Option[OpWorkflow], val asSpark: Bo
    */
   final def load(path: String, modelStagingDir: String = WorkflowFileReader.modelStagingDir): OpWorkflowModel = {
     implicit val conf = new Configuration()
-    val localPath = new Path(modelStagingDir)
     val localFileSystem = FileSystem.getLocal(conf)
+    val localPath = localFileSystem.makeQualified(new Path(modelStagingDir))
     localFileSystem.delete(localPath, true)
 
     val savePath = new Path(path)
     val remoteFileSystem = savePath.getFileSystem(conf)
-
-    val zipPath = new Path(localPath, WorkflowFileReader.zipModel)
-
-    remoteFileSystem.copyToLocalFile(savePath, zipPath)
-
+    val zipDir = new Path(localPath, WorkflowFileReader.zipModel)
+    remoteFileSystem.copyToLocalFile(savePath, zipDir)
+
+    // New serialization:
+    //  remote: savePath (dir) -> Model.zip (file)
+    //  local:  Model.zip (dir) -> Model.zip (file)
+    // Old serialization:
+    //  remote: savePath (dir)
+    //  local:  Model.zip (dir)
     val modelDir = new Path(localPath, WorkflowFileReader.rawModel)
-    val fileToLoad = Try {
-      val zipFile = new File(zipPath.toString)
-      val subZip = // TODO figure out why it puts the files like this
-        if (zipFile.isDirectory) new File(zipFile, WorkflowFileReader.zipModel)
-        else zipFile
-      ZipUtil.unpack(subZip, new File(modelDir.toString))
-    } match { // For backwards compatibility since old models will not be zipped
-      case Success(_) => modelDir.toString
-      case Failure(_) => zipPath.toString
-    }
-
-    val model = Try(WorkflowFileReader.loadFile(OpWorkflowModelReadWriteShared.jsonPath(fileToLoad)))
-      .flatMap(loadJson(_, path = fileToLoad)) match {
-      case Failure(error) => throw new RuntimeException(s"Failed to load Workflow from path '$path'", error)
+    val modelPath = Try {
+      localFileSystem.open(new Path(zipDir, WorkflowFileReader.zipModel))
+    }.map { inputStream =>
+      try {
+        ZipUtil.unpack(inputStream, new File(modelDir.toUri.getPath))
+        modelDir.toString
+      } finally inputStream.close()
+    }.getOrElse(zipDir.toString)
+
+    val model = Try(
+      WorkflowFileReader.loadFile(OpWorkflowModelReadWriteShared.jsonPath(modelPath))
+    ).flatMap(loadJson(_, path = modelPath)) match {
+      case Failure(error) =>
+        throw new RuntimeException(s"Failed to load Workflow from path '$path'", error)
       case Success(wf) => wf
     }
+
     localFileSystem.delete(localPath, true)
     model
   }
@@ -260,7 +265,6 @@ class OpWorkflowModelReader(val workflowOpt: Option[OpWorkflow], val asSpark: Bo
 }
 
 private object WorkflowFileReader {
-
   val rawModel = "rawModel"
   val zipModel = "Model.zip"
   def modelStagingDir: String = s"modelStagingDir/model-${System.currentTimeMillis}"
@@ -286,14 +290,16 @@ private object WorkflowFileReader {
   }
 
   private def readAsString(path: Path)(implicit conf: Configuration): String = {
-    val fs = path.getFileSystem(conf)
     val codecFactory = new CompressionCodecFactory(conf)
     val codec = Option(codecFactory.getCodec(path))
-    val in = fs.open(path)
-    val read = codec.map( c => Source.fromInputStream(c.createInputStream(in)).mkString )
-      .getOrElse( IOUtils.toString(in, "UTF-8") )
-    in.close()
-    read
+    val in = FileSystem.getLocal(conf).open(path)
+    try {
+      val read = codec.map(c => Source.fromInputStream(c.createInputStream(in)).mkString)
+        .getOrElse(IOUtils.toString(in, "UTF-8"))
+      read
+    } finally {
+      in.close()
+    }
   }
 }
 

core/src/main/scala/com/salesforce/op/OpWorkflowModelWriter.scala

@@ -38,7 +38,7 @@ import com.salesforce.op.stages.{OPStage, OpPipelineStageWriter}
 import com.salesforce.op.utils.spark.{JobGroupUtil, OpStep}
 import enumeratum._
 import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.fs.{Path, RawLocalFileSystem}
+import org.apache.hadoop.fs.{FileSystem, Path}
 import org.apache.hadoop.io.compress.GzipCodec
 import org.apache.spark.ml.util.MLWriter
 import org.json4s.JsonAST.{JArray, JObject, JString}
@@ -193,8 +193,6 @@ private[op] object OpWorkflowModelReadWriteShared {
  * Writes the OpWorkflowModel into a specified path
  */
 object OpWorkflowModelWriter {
-  val localFileSystem = new RawLocalFileSystem()
-
   /**
    * Save [[OpWorkflowModel]] to path
    *
@@ -211,7 +209,7 @@ object OpWorkflowModelWriter {
   ): Unit = {
     val localPath = new Path(modelStagingDir)
     val conf = new Configuration()
-
+    val localFileSystem = FileSystem.getLocal(conf)
     if (overwrite) localFileSystem.delete(localPath, true)
     val raw = new Path(modelStagingDir, WorkflowFileReader.rawModel)
 
@@ -221,7 +219,7 @@ object OpWorkflowModelWriter {
     val compressed = new Path(modelStagingDir, WorkflowFileReader.zipModel)
     ZipUtil.pack(new File(raw.toString), new File(compressed.toString))
 
-    val finalPath = new Path(path)
+    val finalPath = new Path(path, WorkflowFileReader.zipModel)
     val destinationFileSystem = finalPath.getFileSystem(conf)
     destinationFileSystem.moveFromLocalFile(compressed, finalPath)
   }

core/src/test/scala/com/salesforce/op/OpWorkflowRunnerTest.scala

@@ -226,7 +226,7 @@ class OpWorkflowRunnerTest extends AsyncFlatSpec with PassengerSparkFixtureTest
       outFile.exists shouldBe true
       val dirFile = if (outFile.getAbsolutePath.endsWith("/model")) {
         val unpacked = new File(outFile.getAbsolutePath + "Unpacked")
-        ZipUtil.unpack(outFile, unpacked)
+        ZipUtil.unpack(new File(outFile, "Model.zip"), unpacked)
         unpacked
       } else outFile
       dirFile.isDirectory shouldBe true

features/src/main/scala/com/salesforce/op/stages/SparkStageParam.scala

@@ -34,11 +34,12 @@ import com.salesforce.op.stages.sparkwrappers.generic.SparkWrapperParams
 import ml.combust.bundle.BundleFile
 import ml.combust.bundle.dsl.Bundle
 import ml.combust.bundle.serializer.SerializationFormat
-import ml.combust.mleap.runtime.MleapSupport._
+import ml.combust.mleap.runtime.MleapSupport.MleapBundleFileOps
 import ml.combust.mleap.runtime.frame.{Transformer => MLeapTransformer}
 import ml.combust.mleap.spark.SparkSupport._
 import ml.combust.mleap.xgboost.runtime.bundle.ops.{XGBoostClassificationOp, XGBoostRegressionOp}
-import org.apache.hadoop.fs.{Path, RawLocalFileSystem}
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.{FileSystem, Path}
 import org.apache.spark.ml.bundle.SparkBundleContext
 import org.apache.spark.ml.param.{Param, ParamPair, Params}
 import org.apache.spark.ml.util.{Identifiable, MLReader, MLWritable}
@@ -62,8 +63,6 @@ class SparkStageParam[S <: PipelineStage with Params]
 
   import SparkStageParam._
 
-  @transient val rawLocalFileSystem = new RawLocalFileSystem()
-
   /**
    * Spark stage saving path
    */
@@ -92,7 +91,8 @@ class SparkStageParam[S <: PipelineStage with Params]
     def json(className: String, uid: String) = compact(render(("className" -> className) ~ ("uid" -> uid)))
     (sparkStage, savePath, sbc) match {
       case (Some(stage), Some(path), Some(c)) =>
-        val stagePath = rawLocalFileSystem.makeQualified(new Path(path, stage.uid))
+        val stagePath = SparkStageParam.localFileSystem.makeQualified(new Path(path, stage.uid))
+
         for {bundle <- managed(BundleFile(stagePath.toUri))} {
           stage.asInstanceOf[Transformer].writeBundle.format(SerializationFormat.Json).save(bundle)(c)
             .getOrElse(throw new RuntimeException(s"Failed to write $stage to $path with context $c"))
@@ -166,7 +166,7 @@ class SparkStageParam[S <: PipelineStage with Params]
         None
       case (Some(path), Some(stageUid), asSpark, className) =>
         savePath = Option(path)
-        val stagePath = rawLocalFileSystem.makeQualified(new Path(path, stageUid))
+        val stagePath = SparkStageParam.localFileSystem.makeQualified(new Path(path, stageUid))
         val loaded = for {bundle <- managed(BundleFile(stagePath.toUri))} yield {
           // TODO remove random forest regression when mleap spark deserialization is fixed
           // https://github.com/combust/mleap/issues/721
@@ -191,6 +191,8 @@ object SparkStageParam {
   val NoUID = ""
   val RandomForestRegressor = "org.apache.spark.ml.regression.RandomForestRegressionModel"
 
+  val localFileSystem = FileSystem.getLocal(new Configuration())
+
   def updateParamsMetadataWithPath(jValue: JValue, path: String, asSpark: Boolean): JValue = jValue match {
     case JObject(pairs) => JObject(
       pairs.map {