添加缺失的类

Author: dapeng

…l/sink/clickhouse/ClickhouseDialect.java …sql/launcher/perjob/ConfigParseUtil.javaclickhouse/clickhouse-sink/src/main/java/com/dtstack/flink/sql/sink/clickhouse/ClickhouseDialect.java renamed to launcher/src/main/java/com/dtstack/flink/sql/launcher/perjob/ConfigParseUtil.java clickhouse/clickhouse-sink/src/main/java/com/dtstack/flink/sql/sink/clickhouse/ClickhouseDialect.java renamed to launcher/src/main/java/com/dtstack/flink/sql/launcher/perjob/ConfigParseUtil.java

@@ -16,31 +16,27 @@
  * limitations under the License.
  */
 
-package com.dtstack.flink.sql.sink.clickhouse;
+package com.dtstack.flink.sql.launcher.perjob;
 
-import com.dtstack.flink.sql.sink.rdb.dialect.JDBCDialect;
+import org.apache.commons.io.Charsets;
 
-import java.util.Optional;
+import java.io.UnsupportedEncodingException;
+import java.net.URLDecoder;
+import java.util.Arrays;
+import java.util.List;
 
 /**
- * Date: 2020/1/15
+ * Date: 2019/12/28
  * Company: www.dtstack.com
  * @author maqi
  */
-public class ClickhouseDialect implements JDBCDialect {
+public class ConfigParseUtil {
 
-    @Override
-    public boolean canHandle(String url) {
-        return url.startsWith("jdbc:clickhouse:");
-    }
-
-    @Override
-    public Optional<String> defaultDriverName() {
-        return Optional.of("ru.yandex.clickhouse.ClickHouseDriver");
-    }
-
-    @Override
-    public String getUpdateStatement(String tableName, String[] fieldNames, String[] conditionFields) {
-        throw new RuntimeException("Clickhouse does not support update sql, please remove primary key or use append mode");
+    public static List<String> parsePathFromStr(String pathStr) throws UnsupportedEncodingException {
+        String addjarPath = URLDecoder.decode(pathStr, Charsets.UTF_8.toString());
+        if (addjarPath.length() > 2) {
+            addjarPath = addjarPath.substring(1,addjarPath.length()-1).replace("\"","");
+        }
+        return Arrays.asList(addjarPath.split(","));
     }
 }

launcher/src/main/java/com/dtstack/flink/sql/launcher/perjob/PerJobClusterClientBuilder.java

@@ -38,6 +38,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.File;
+import java.io.UnsupportedEncodingException;
 import java.net.MalformedURLException;
 import java.net.URL;
 import java.util.ArrayList;
@@ -82,7 +83,7 @@ public void init(String yarnConfDir, Configuration flinkConfig, Properties userC
     }
 
     public AbstractYarnClusterDescriptor createPerJobClusterDescriptor(String flinkJarPath, Options launcherOptions, JobGraph jobGraph)
-            throws MalformedURLException {
+            throws MalformedURLException, UnsupportedEncodingException {
 
         String flinkConf = StringUtils.isEmpty(launcherOptions.getFlinkconf()) ? DEFAULT_CONF_DIR : launcherOptions.getFlinkconf();
         AbstractYarnClusterDescriptor clusterDescriptor = getClusterDescriptor(flinkConfig, yarnConf, flinkConf);
@@ -106,13 +107,7 @@ public AbstractYarnClusterDescriptor createPerJobClusterDescriptor(String flinkJ
         } else {
             throw new RuntimeException("The Flink jar path is null");
         }
-        // add  user customized file to shipfile
-        if (!StringUtils.isBlank(launcherOptions.getAddShipfile())) {
-            List<String> paths = ConfigParseUtil.parsePathFromStr(launcherOptions.getAddShipfile());
-            paths.forEach(path -> {
-                shipFiles.add(new File(path));
-            });
-        }
+
         // classpath , all node need contain plugin jar
         String pluginLoadMode = launcherOptions.getPluginLoadMode();
         if (StringUtils.equalsIgnoreCase(pluginLoadMode, EPluginLoadMode.CLASSPATH.name())) {
@@ -125,6 +120,14 @@ public AbstractYarnClusterDescriptor createPerJobClusterDescriptor(String flinkJ
                     + " Currently only classpath and shipfile are supported.");
         }
 
+        // add  user customized file to shipfile
+        if (!StringUtils.isBlank(launcherOptions.getAddShipfile())) {
+            List<String> paths = ConfigParseUtil.parsePathFromStr(launcherOptions.getAddShipfile());
+            paths.forEach(path -> {
+                shipFiles.add(new File(path));
+            });
+        }
+
         clusterDescriptor.addShipFiles(shipFiles);
         clusterDescriptor.setName(launcherOptions.getName());
         String queue = launcherOptions.getQueue();