feat: support spark conf path for sdk (#2180)

Author: tobegit3hub

java/openmldb-jdbc/src/main/java/com/_4paradigm/openmldb/sdk/SdkOption.java

@@ -18,18 +18,20 @@
 
 import lombok.Data;
 
-
 @Data
 public class SdkOption {
-//    options for cluster mode
+    // options for cluster mode
     private String zkCluster;
     private String zkPath;
-//    options for standalone mode
+    private String sparkConfPath = "";
+
+    // options for standalone mode
     private String host;
     private long port;
 
     private long sessionTimeout = 10000;
     private Boolean enableDebug = false;
     private long requestTimeout = 60000;
     private boolean isClusterMode = true;
+
 }

java/openmldb-jdbc/src/main/java/com/_4paradigm/openmldb/sdk/impl/SqlClusterExecutor.java

@@ -69,6 +69,7 @@ public SqlClusterExecutor(SdkOption option, String libraryPath) throws SqlExcept
             sqlOpt.setZk_path(option.getZkPath());
             sqlOpt.setEnable_debug(option.getEnableDebug());
             sqlOpt.setRequest_timeout(option.getRequestTimeout());
+            sqlOpt.setSpark_conf_path(option.getSparkConfPath());
             this.sqlRouter = sql_router_sdk.NewClusterSQLRouter(sqlOpt);
             sqlOpt.delete();
         } else {

python/openmldb/sdk/sdk.py

@@ -32,10 +32,11 @@
 
 
 class OpenMLDBClusterSdkOptions(object):
-    def __init__(self, zk_cluster, zk_path, session_timeout=3000):
+    def __init__(self, zk_cluster, zk_path, session_timeout=3000, spark_conf_path=""):
         self.zk_cluster = zk_cluster
         self.zk_path = zk_path
         self.zk_session_timeout = session_timeout
+        self.spark_conf_path = spark_conf_path
 
 
 class OpenMLDBStandaloneSdkOptions(object):

src/cmd/sql_cmd.h

@@ -35,6 +35,7 @@
 DEFINE_bool(interactive, true, "Set the interactive");
 DEFINE_string(database, "", "Set database");
 DECLARE_string(cmd);
+DEFINE_string(spark_conf, "", "The config file of Spark job");
 
 // cluster mode
 DECLARE_string(zk_cluster);
@@ -209,6 +210,7 @@ bool InitClusterSDK() {
     copt.zk_session_timeout = FLAGS_zk_session_timeout;
     copt.zk_log_level = FLAGS_zk_log_level;
     copt.zk_log_file = FLAGS_zk_log_file;
+
     cs = new ::openmldb::sdk::ClusterSDK(copt);
     if (!cs->Init()) {
         std::cout << "ERROR: Failed to connect to db" << std::endl;
@@ -220,6 +222,9 @@ bool InitClusterSDK() {
         return false;
     }
     sr->SetInteractive(FLAGS_interactive);
+
+    sr->GetSqlRouterOptions().spark_conf_path = FLAGS_spark_conf;
+
     return true;
 }
 

src/sdk/sql_cluster_router.cc

@@ -50,7 +50,6 @@
 
 DECLARE_int32(request_timeout_ms);
 DECLARE_string(bucket_size);
-DEFINE_string(spark_conf, "", "The config file of Spark job");
 DECLARE_uint32(replica_num);
 
 namespace openmldb {
@@ -2442,7 +2441,7 @@ std::shared_ptr<hybridse::sdk::ResultSet> SQLClusterRouter::ExecuteSQL(const std
             } else {
                 ::openmldb::taskmanager::JobInfo job_info;
                 std::map<std::string, std::string> config;
-                ReadSparkConfFromFile(FLAGS_spark_conf, &config);
+                ReadSparkConfFromFile(options_.spark_conf_path, &config);
                 auto base_status = ExportOfflineData(sql, config, db, is_sync_job, offline_job_timeout, &job_info);
                 if (base_status.OK()) {
                     *status = {};
@@ -2473,7 +2472,7 @@ std::shared_ptr<hybridse::sdk::ResultSet> SQLClusterRouter::ExecuteSQL(const std
                 // Handle in cluster mode
                 ::openmldb::taskmanager::JobInfo job_info;
                 std::map<std::string, std::string> config;
-                ReadSparkConfFromFile(FLAGS_spark_conf, &config);
+                ReadSparkConfFromFile(options_.spark_conf_path, &config);
 
                 ::openmldb::base::Status base_status;
                 if (is_online_mode) {
@@ -2518,7 +2517,7 @@ std::shared_ptr<hybridse::sdk::ResultSet> SQLClusterRouter::ExecuteOfflineQuery(
                                                                                 bool is_sync_job, int job_timeout,
                                                                                 ::hybridse::sdk::Status* status) {
     std::map<std::string, std::string> config;
-    ReadSparkConfFromFile(FLAGS_spark_conf, &config);
+    ReadSparkConfFromFile(options_.spark_conf_path, &config);
 
     if (is_sync_job) {
         // Run offline sql and wait to get output
@@ -3721,15 +3720,15 @@ std::shared_ptr<hybridse::sdk::ResultSet> SQLClusterRouter::ExecuteShowTableStat
     return ResultSetSQL::MakeResultSet(GetTableStatusSchema(), data, status);
 }
 
-void SQLClusterRouter::ReadSparkConfFromFile(std::string conf_file, std::map<std::string, std::string>* config) {
-    if (!conf_file.empty()) {
+void SQLClusterRouter::ReadSparkConfFromFile(std::string conf_file_path, std::map<std::string, std::string>* config) {
+    if (!conf_file_path.empty()) {
         boost::property_tree::ptree pt;
 
         try {
-            boost::property_tree::ini_parser::read_ini(FLAGS_spark_conf, pt);
-            LOG(INFO) << "Load Spark conf file: " << conf_file;
+            boost::property_tree::ini_parser::read_ini(conf_file_path, pt);
+            LOG(INFO) << "Load Spark conf file: " << conf_file_path;
         } catch (...) {
-            LOG(WARNING) << "Fail to load Spark conf file: " << conf_file;
+            LOG(WARNING) << "Fail to load Spark conf file: " << conf_file_path;
             return;
         }
 

src/sdk/sql_cluster_router.h

@@ -298,6 +298,10 @@ class SQLClusterRouter : public SQLRouter {
 
     void ReadSparkConfFromFile(std::string conf_file, std::map<std::string, std::string>* config);
 
+    SQLRouterOptions GetSqlRouterOptions() {
+        return options_;
+    }
+
  private:
     bool IsSyncJob();
     // get job timeout from the session variables, we will use the timeout when sending requests to the taskmanager

src/sdk/sql_router.h

@@ -46,6 +46,7 @@ struct SQLRouterOptions : BasicRouterOptions {
     std::string zk_cluster;
     std::string zk_path;
     uint32_t zk_session_timeout = 2000;
+    std::string spark_conf_path;
 };
 
 struct StandaloneOptions : BasicRouterOptions {