Support spark sql

flink-platform-web/src/main/java/com/flink/platform/web/command/sql/SqlCommandExecutor.java

@@ -2,12 +2,15 @@
 
 import com.flink.platform.common.enums.JobType;
 import com.flink.platform.dao.entity.Datasource;
+import com.flink.platform.dao.entity.JobRunInfo;
 import com.flink.platform.dao.entity.result.JobCallback;
 import com.flink.platform.dao.service.DatasourceService;
+import com.flink.platform.dao.service.JobRunInfoService;
 import com.flink.platform.web.command.AbstractTask;
 import com.flink.platform.web.command.CommandExecutor;
 import com.flink.platform.web.command.JobCommand;
 import jakarta.annotation.Nonnull;
+import jakarta.annotation.Resource;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
@@ -24,6 +27,9 @@ public class SqlCommandExecutor implements CommandExecutor {
     @Autowired
     private DatasourceService datasourceService;
 
+    @Resource
+    private JobRunInfoService jobRunInfoService;
+
     @Override
     public boolean isSupported(JobType jobType) {
         return jobType == CLICKHOUSE_SQL || jobType == MYSQL_SQL || jobType == HIVE_SQL;
@@ -34,10 +40,10 @@ public boolean isSupported(JobType jobType) {
     public JobCallback execCommand(@Nonnull JobCommand command) throws Exception {
         SqlCommand sqlCommand = (SqlCommand) command;
         Datasource datasource = datasourceService.getById(sqlCommand.getDsId());
-        SqlTask task = new SqlTask(sqlCommand.getJobRunId(), sqlCommand.getSqls(), datasource);
+        JobRunInfo jobRunInfo = jobRunInfoService.getById(sqlCommand.getJobRunId());
+        SqlTask task = new SqlTask(jobRunInfo, sqlCommand.getSqls(), datasource);
         sqlCommand.setTask(task);
         task.run();
-
         return task.buildResult();
     }
 

flink-platform-web/src/main/java/com/flink/platform/web/command/sql/SqlTask.java

@@ -4,6 +4,7 @@
 import com.flink.platform.common.util.ExceptionUtil;
 import com.flink.platform.common.util.JsonUtil;
 import com.flink.platform.dao.entity.Datasource;
+import com.flink.platform.dao.entity.JobRunInfo;
 import com.flink.platform.dao.entity.result.JobCallback;
 import com.flink.platform.web.command.AbstractTask;
 import jakarta.annotation.Nonnull;
@@ -35,6 +36,8 @@ public class SqlTask extends AbstractTask {
 
     private final Datasource datasource;
 
+    private final JobRunInfo jobRunInfo;
+
     private final List<String> sqlList;
 
     private volatile Statement statement;
@@ -45,17 +48,18 @@ public class SqlTask extends AbstractTask {
 
     private String sqlResult;
 
-    public SqlTask(long jobRunId, @Nonnull List<String> sqlList, @Nonnull Datasource datasource) {
-        super(jobRunId);
+    public SqlTask(JobRunInfo jobRunInfo, @Nonnull List<String> sqlList, @Nonnull Datasource datasource) {
+        super(checkNotNull(jobRunInfo).getId());
         this.sqlList = checkNotNull(sqlList);
         this.datasource = checkNotNull(datasource);
+        this.jobRunInfo = checkNotNull(jobRunInfo);
     }
 
     @Override
     public void run() throws Exception {
         Exception exception = null;
         List<Map<String, Object>> dataList = new ArrayList<>();
-        try (Connection connection = createConnection(datasource.getType(), datasource.getParams());
+        try (Connection connection = createConnection(datasource.getType(), datasource.getParams(), jobRunInfo);
                 Statement stmt = connection.createStatement()) {
             this.statement = stmt;
 

flink-platform-web/src/main/java/com/flink/platform/web/service/ReactiveService.java

@@ -103,7 +103,7 @@ public ReactiveDataVo execSql(String execId, JobInfo jobInfo, Datasource datasou
         }
 
         String statement = sqls.get(0).toSqlString();
-        try (Connection connection = createConnection(datasource.getType(), datasource.getParams());
+        try (Connection connection = createConnection(datasource.getType(), datasource.getParams(), null);
                 Statement stmt = connection.createStatement()) {
             String[] columnNames;
             List<Object[]> dataList = new ArrayList<>();

flink-platform-web/src/main/java/com/flink/platform/web/util/JdbcUtil.java

@@ -1,20 +1,26 @@
 package com.flink.platform.web.util;
 
 import com.flink.platform.common.enums.DbType;
+import com.flink.platform.dao.entity.JobRunInfo;
 import com.flink.platform.dao.entity.ds.DatasourceParam;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.collections4.MapUtils;
+import org.apache.commons.lang3.StringUtils;
 
 import java.sql.Array;
 import java.sql.Connection;
 import java.sql.DriverManager;
 import java.sql.SQLException;
+import java.util.Objects;
 import java.util.Properties;
 
 /** Jdbc util. */
+@Slf4j
 public class JdbcUtil {
 
     /** create datasource. */
-    public static Connection createConnection(DbType dbType, DatasourceParam params) throws Exception {
+    public static Connection createConnection(DbType dbType, DatasourceParam params, JobRunInfo jobRunInfo)
+            throws Exception {
         Class.forName(dbType.getDriver());
         Properties properties = new Properties();
 
@@ -36,7 +42,15 @@ public static Connection createConnection(DbType dbType, DatasourceParam params)
             properties.putAll(params.getProperties());
         }
 
-        return DriverManager.getConnection(params.getUrl(), properties);
+        String url = params.getUrl();
+        if (Objects.equals(dbType, DbType.HIVE)) {
+            url = addAppName(params.getUrl(), jobRunInfo);
+        }
+        log.info("create connection url: {}", url);
+
+        // avoid hive connection creation timeout.
+        DriverManager.setLoginTimeout(600);
+        return DriverManager.getConnection(url, properties);
     }
 
     public static Object toJavaObject(Object dbObject, DbType dbType) throws SQLException {
@@ -59,4 +73,25 @@ public static Object toJavaObject(Object dbObject, DbType dbType) throws SQLExce
                 return dbObject;
         }
     }
+
+    private static String addAppName(String url, JobRunInfo jobRunInfo) {
+        if (StringUtils.isBlank(url) || Objects.isNull(jobRunInfo)) {
+            return url;
+        }
+        String sparkAppName = createAppName(jobRunInfo);
+        if (StringUtils.contains(url, "#")) {
+            return url + ";spark.app.name=" + sparkAppName;
+        }
+        return url + "#spark.app.name=" + sparkAppName;
+    }
+
+    private static String createAppName(JobRunInfo jobRun) {
+        String jobName = jobRun.getName().replaceAll("\\s+", "");
+        return String.join(
+                "-",
+                jobRun.getExecMode().name(),
+                jobRun.getJobCode() + "_" + jobRun.getFlowRunId(),
+                jobName,
+                String.valueOf(jobRun.getUserId()));
+    }
 }

pom.xml

@@ -231,10 +231,6 @@
 						<artifactId>jline</artifactId>
 						<groupId>jline</groupId>
 					</exclusion>
-					<exclusion>
-						<groupId>org.apache.zookeeper</groupId>
-						<artifactId>zookeeper</artifactId>
-					</exclusion>
 					<exclusion>
 						<artifactId>curator-recipes</artifactId>
 						<groupId>org.apache.curator</groupId>