Pass sql as job info args

4paradigm · Nov 29, 2023 · 0c9f9e8 · 0c9f9e8
1 parent 71cf61e
commit 0c9f9e8
Show file tree

Hide file tree

Showing 4 changed files with 28 additions and 13 deletions.
diff --git a/java/openmldb-batch/src/main/scala/com/_4paradigm/openmldb/batch/api/OpenmldbSession.scala b/java/openmldb-batch/src/main/scala/com/_4paradigm/openmldb/batch/api/OpenmldbSession.scala
@@ -174,6 +174,8 @@ class OpenmldbSession {
    * @return
    */
   def openmldbSql(sqlText: String): OpenmldbDataframe = {
+    logger.info("Try to execute OpenMLDB SQL: " + sqlText)
+
     if (config.enableSparksql) {
       return OpenmldbDataframe(this, sparksql(sqlText))
     }
@@ -323,7 +325,7 @@ class OpenmldbSession {
             }
           } catch {
             case e: Exception => {
-              logger.warn(s"Fail to register table $dbName.$tableName " + ExceptionUtils.getStackTrace(e))
+              logger.warn(s"Fail to register table $dbName.$tableName, exception: " + e.getMessage)
             }
           }
         }

diff --git a/...kmanager/src/main/scala/com/_4paradigm/openmldb/taskmanager/OpenmldbBatchjobManager.scala b/...kmanager/src/main/scala/com/_4paradigm/openmldb/taskmanager/OpenmldbBatchjobManager.scala
@@ -57,10 +57,10 @@ object OpenmldbBatchjobManager {
 
     if (TaskManagerConfig.isK8s) {
       val args = List(sql)
-      K8sJobManager.submitSparkJob(jobType, mainClass, args, tempSqlFile.getAbsolutePath, sparkConf.asScala.toMap,
+      K8sJobManager.submitSparkJob(jobType, mainClass, args, sql, tempSqlFile.getAbsolutePath, sparkConf.asScala.toMap,
         defaultDb)
     } else {
-      SparkJobManager.submitSparkJob(jobType, mainClass, args, tempSqlFile.getAbsolutePath,
+      SparkJobManager.submitSparkJob(jobType, mainClass, args, sql, tempSqlFile.getAbsolutePath,
         sparkConf.asScala.toMap, defaultDb, blocking = true)
     }
   }
@@ -73,11 +73,11 @@ object OpenmldbBatchjobManager {
 
     if (TaskManagerConfig.isK8s) {
       val args = List(sql)
-      K8sJobManager.submitSparkJob(jobType, mainClass, args, tempSqlFile.getAbsolutePath, sparkConf.asScala.toMap,
+      K8sJobManager.submitSparkJob(jobType, mainClass, args, sql, tempSqlFile.getAbsolutePath, sparkConf.asScala.toMap,
         defaultDb)
     } else {
       val args = List(tempSqlFile.getAbsolutePath)
-      SparkJobManager.submitSparkJob(jobType, mainClass, args, tempSqlFile.getAbsolutePath, sparkConf.asScala.toMap,
+      SparkJobManager.submitSparkJob(jobType, mainClass, args, sql, tempSqlFile.getAbsolutePath, sparkConf.asScala.toMap,
         defaultDb)
     }
   }
@@ -90,11 +90,11 @@ object OpenmldbBatchjobManager {
 
     if (TaskManagerConfig.isK8s) {
       val args = List(sql)
-      K8sJobManager.submitSparkJob(jobType, mainClass, args, tempSqlFile.getAbsolutePath, sparkConf.asScala.toMap,
+      K8sJobManager.submitSparkJob(jobType, mainClass, args, sql, tempSqlFile.getAbsolutePath, sparkConf.asScala.toMap,
         defaultDb)
     } else {
       val args = List(tempSqlFile.getAbsolutePath)
-      SparkJobManager.submitSparkJob(jobType, mainClass, args, tempSqlFile.getAbsolutePath, sparkConf.asScala.toMap,
+      SparkJobManager.submitSparkJob(jobType, mainClass, args, sql, tempSqlFile.getAbsolutePath, sparkConf.asScala.toMap,
         defaultDb)
     }
   }
@@ -107,11 +107,11 @@ object OpenmldbBatchjobManager {
 
     if (TaskManagerConfig.isK8s) {
       val args = List(sql)
-      K8sJobManager.submitSparkJob(jobType, mainClass, args, tempSqlFile.getAbsolutePath, sparkConf.asScala.toMap,
+      K8sJobManager.submitSparkJob(jobType, mainClass, args, sql, tempSqlFile.getAbsolutePath, sparkConf.asScala.toMap,
         defaultDb)
     } else {
       val args = List(tempSqlFile.getAbsolutePath)
-      SparkJobManager.submitSparkJob(jobType, mainClass, args, tempSqlFile.getAbsolutePath, sparkConf.asScala.toMap,
+      SparkJobManager.submitSparkJob(jobType, mainClass, args, sql, tempSqlFile.getAbsolutePath, sparkConf.asScala.toMap,
         defaultDb)
     }
   }
@@ -124,11 +124,11 @@ object OpenmldbBatchjobManager {
 
     if (TaskManagerConfig.isK8s) {
       val args = List(sql)
-      K8sJobManager.submitSparkJob(jobType, mainClass, args, tempSqlFile.getAbsolutePath, sparkConf.asScala.toMap,
+      K8sJobManager.submitSparkJob(jobType, mainClass, args, sql, tempSqlFile.getAbsolutePath, sparkConf.asScala.toMap,
         defaultDb)
     } else {
       val args = List(tempSqlFile.getAbsolutePath)
-      SparkJobManager.submitSparkJob(jobType, mainClass, args, tempSqlFile.getAbsolutePath, sparkConf.asScala.toMap,
+      SparkJobManager.submitSparkJob(jobType, mainClass, args, sql, tempSqlFile.getAbsolutePath, sparkConf.asScala.toMap,
         defaultDb)
     }
   }

diff --git a/...db-taskmanager/src/main/scala/com/_4paradigm/openmldb/taskmanager/k8s/K8sJobManager.scala b/...db-taskmanager/src/main/scala/com/_4paradigm/openmldb/taskmanager/k8s/K8sJobManager.scala
@@ -44,12 +44,18 @@ object K8sJobManager {
 
   def submitSparkJob(jobType: String, mainClass: String,
     args: List[String] = List(),
+    sql: String = "",
     localSqlFile: String = "",
     sparkConf: Map[String, String] = Map(),
     defaultDb: String = "",
     blocking: Boolean = false): JobInfo = {
 
-    val jobInfo = JobInfoManager.createJobInfo(jobType, args, sparkConf)
+    val jobInfoArgs = if (sql.nonEmpty) {
+      List(sql)
+    } else {
+      args
+    }
+    val jobInfo = JobInfoManager.createJobInfo(jobType, jobInfoArgs, sparkConf)
 
     val jobName = getK8sJobName(jobInfo.getId)
     jobInfo.setApplicationId(jobName)

diff --git a/...askmanager/src/main/scala/com/_4paradigm/openmldb/taskmanager/spark/SparkJobManager.scala b/...askmanager/src/main/scala/com/_4paradigm/openmldb/taskmanager/spark/SparkJobManager.scala
@@ -77,11 +77,18 @@ object SparkJobManager {
 
   def submitSparkJob(jobType: String, mainClass: String,
                      args: List[String] = List(),
+                     sql: String = "",
                      localSqlFile: String = "",
                      sparkConf: Map[String, String] = Map(),
                      defaultDb: String = "",
                      blocking: Boolean = false): JobInfo = {
-    val jobInfo = JobInfoManager.createJobInfo(jobType, args, sparkConf)
+
+    val jobInfoArgs = if (sql.nonEmpty) {
+      List(sql)
+    } else {
+      args
+    }
+    val jobInfo = JobInfoManager.createJobInfo(jobType, jobInfoArgs, sparkConf)
 
     // Submit Spark application with SparkLauncher
     val launcher = createSparkLauncher(mainClass)