feat: read spark.master and set correct sql file path (#1996)

* Read spark.master and set correct sql file path

* Reuse util methods for all batch job

Author: tobegit3hub

java/openmldb-batchjob/src/main/scala/com/_4paradigm/openmldb/batchjob/ExportOfflineData.scala

@@ -16,25 +16,19 @@
 
 package com._4paradigm.openmldb.batchjob
 
-import com._4paradigm.openmldb.batch.api.OpenmldbSession
-import org.apache.spark.SparkFiles
+import com._4paradigm.openmldb.batchjob.util.OpenmldbJobUtil
 import org.apache.spark.sql.SparkSession
 
 object ExportOfflineData {
-  def main(args: Array[String]): Unit = {
-    if (args.length < 1) {
-      throw new Exception(s"Require args: sql but get args: ${args.mkString(",")}")
-    }
 
+  def main(args: Array[String]): Unit = {
+    OpenmldbJobUtil.checkOneSqlArgument(args)
     exportOfflineData(args(0))
   }
 
   def exportOfflineData(sqlFilePath: String): Unit = {
-    val sess = new OpenmldbSession(SparkSession.builder().getOrCreate())
-
-    val sqlText = scala.io.Source.fromFile(SparkFiles.get(sqlFilePath)).mkString
-    sess.sql(sqlText)
-
-    sess.close()
+    val spark = SparkSession.builder().getOrCreate()
+    OpenmldbJobUtil.runOpenmldbSql(spark, sqlFilePath)
   }
+
 }

java/openmldb-batchjob/src/main/scala/com/_4paradigm/openmldb/batchjob/ImportOfflineData.scala

@@ -16,27 +16,19 @@
 
 package com._4paradigm.openmldb.batchjob
 
-import com._4paradigm.openmldb.batch.api.OpenmldbSession
-import org.apache.spark.SparkFiles
+import com._4paradigm.openmldb.batchjob.util.OpenmldbJobUtil
 import org.apache.spark.sql.SparkSession
 
 object ImportOfflineData {
 
   def main(args: Array[String]): Unit = {
-    if (args.length < 1) {
-      throw new Exception(s"Require args: sql but get args: ${args.mkString(",")}")
-    }
-
+    OpenmldbJobUtil.checkOneSqlArgument(args)
     importOfflineData(args(0))
   }
 
   def importOfflineData(sqlFilePath: String): Unit = {
-    val sess = new OpenmldbSession(SparkSession.builder().getOrCreate())
-
-    val sqlText = scala.io.Source.fromFile(SparkFiles.get(sqlFilePath)).mkString
-    sess.sql(sqlText)
-
-    sess.close()
+    val spark = SparkSession.builder().getOrCreate()
+    OpenmldbJobUtil.runOpenmldbSql(spark, sqlFilePath)
   }
 
 }

java/openmldb-batchjob/src/main/scala/com/_4paradigm/openmldb/batchjob/ImportOnlineData.scala

@@ -16,27 +16,19 @@
 
 package com._4paradigm.openmldb.batchjob
 
-import com._4paradigm.openmldb.batch.api.OpenmldbSession
-import org.apache.spark.SparkFiles
+import com._4paradigm.openmldb.batchjob.util.OpenmldbJobUtil
 import org.apache.spark.sql.SparkSession
 
 object ImportOnlineData {
 
   def main(args: Array[String]): Unit = {
-    if (args.length < 1) {
-      throw new Exception(s"Require args: sql but get args: ${args.mkString(",")}")
-    }
-
+    OpenmldbJobUtil.checkOneSqlArgument(args)
     importOnlineData(args(0))
   }
 
   def importOnlineData(sqlFilePath: String): Unit = {
-    val sess = new OpenmldbSession(SparkSession.builder().config("openmldb.loaddata.mode", "online").getOrCreate())
-
-    val sqlText = scala.io.Source.fromFile(SparkFiles.get(sqlFilePath)).mkString
-    sess.sql(sqlText)
-
-    sess.close()
+    val spark = SparkSession.builder().config("openmldb.loaddata.mode", "online").getOrCreate()
+    OpenmldbJobUtil.runOpenmldbSql(spark, sqlFilePath)
   }
 
 }

java/openmldb-batchjob/src/main/scala/com/_4paradigm/openmldb/batchjob/RunBatchAndShow.scala

@@ -17,25 +17,22 @@
 package com._4paradigm.openmldb.batchjob
 
 import com._4paradigm.openmldb.batch.api.OpenmldbSession
-import org.apache.spark.SparkFiles
+import com._4paradigm.openmldb.batchjob.util.OpenmldbJobUtil
 import org.apache.spark.sql.SparkSession
 
 object RunBatchAndShow {
 
   def main(args: Array[String]): Unit = {
-    if (args.length < 1) {
-      throw new Exception(s"Require args: sql but get args: ${args.mkString(",")}")
-    }
-
+    OpenmldbJobUtil.checkOneSqlArgument(args)
     runBatchSql(args(0))
   }
 
   def runBatchSql(sqlFilePath: String): Unit = {
-    val sess = new OpenmldbSession(SparkSession.builder().getOrCreate())
+    val spark = SparkSession.builder().getOrCreate()
+    val sqlText = OpenmldbJobUtil.getSqlFromFile(spark, sqlFilePath)
 
-    val sqlText = scala.io.Source.fromFile(SparkFiles.get(sqlFilePath)).mkString
+    val sess = new OpenmldbSession(spark)
     sess.sql(sqlText).show()
-
     sess.close()
   }
 

java/openmldb-batchjob/src/main/scala/com/_4paradigm/openmldb/batchjob/RunBatchSql.scala

@@ -17,25 +17,23 @@
 package com._4paradigm.openmldb.batchjob
 
 import com._4paradigm.openmldb.batch.api.OpenmldbSession
+import com._4paradigm.openmldb.batchjob.util.OpenmldbJobUtil
 import org.apache.spark.SparkFiles
 import org.apache.spark.sql.SparkSession
 
 object RunBatchSql {
 
   def main(args: Array[String]): Unit = {
-    if (args.length < 1) {
-      throw new Exception(s"Require args: sql but get args: ${args.mkString(",")}")
-    }
-
+    OpenmldbJobUtil.checkOneSqlArgument(args)
     runBatchSql(args(0))
   }
 
   def runBatchSql(sqlFilePath: String): Unit = {
-    val sess = new OpenmldbSession(SparkSession.builder().getOrCreate())
+    val spark = SparkSession.builder().getOrCreate()
+    val sqlText = OpenmldbJobUtil.getSqlFromFile(spark, sqlFilePath)
 
-    val sqlText = scala.io.Source.fromFile(SparkFiles.get(sqlFilePath)).mkString
+    val sess = new OpenmldbSession(spark)
     sess.sql(sqlText).show()
-
     sess.close()
   }
 

java/openmldb-batchjob/src/main/scala/com/_4paradigm/openmldb/batchjob/util/OpenmldbJobUtil.scala

@@ -0,0 +1,56 @@
+/*
+ * Copyright 2021 4Paradigm
+ *
+ * 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.
+ */
+
+package com._4paradigm.openmldb.batchjob.util
+
+import com._4paradigm.openmldb.batch.api.OpenmldbSession
+import org.apache.spark.SparkFiles
+import org.apache.spark.sql.SparkSession
+import scala.reflect.io.File
+
+object OpenmldbJobUtil {
+
+  def checkOneSqlArgument(args: Array[String]): Unit = {
+    if (args.length != 1) {
+      throw new Exception(s"Require args of sql but get args: ${args.mkString(",")}")
+    }
+  }
+
+  def getSqlFromFile(spark: SparkSession, sqlFilePath: String): String = {
+    val sparkMaster = spark.conf.get("spark.master")
+
+    val actualSqlFilePath = if (sparkMaster.equals("local")) {
+      SparkFiles.get(sqlFilePath)
+    } else {
+      sqlFilePath
+    }
+
+    if (!File(actualSqlFilePath).exists) {
+      throw new Exception("SQL file does not exist in " + actualSqlFilePath)
+    }
+
+    scala.io.Source.fromFile(actualSqlFilePath).mkString
+  }
+
+  def runOpenmldbSql(spark: SparkSession, sqlFilePath: String): Unit = {
+    val sqlText = OpenmldbJobUtil.getSqlFromFile(spark, sqlFilePath)
+
+    val sess = new OpenmldbSession(spark)
+    sess.sql(sqlText)
+    sess.close()
+  }
+
+}