DTStack
diff --git a/‎docs/generalconfig.md‎
Lines changed: 2 additions & 6 deletions b/‎docs/generalconfig.md‎
Lines changed: 2 additions & 6 deletions
diff --git a/‎docs/offline/reader/mongodbreader.md‎
Lines changed: 1 addition & 0 deletions b/‎docs/offline/reader/mongodbreader.md‎
Lines changed: 1 addition & 0 deletions
diff --git a/‎docs/offline/reader/sqlserverreader.md‎
Lines changed: 1 addition & 0 deletions b/‎docs/offline/reader/sqlserverreader.md‎
Lines changed: 1 addition & 0 deletions
diff --git a/‎docs/quickstart.md‎
Lines changed: 1 addition & 1 deletion b/‎docs/quickstart.md‎
Lines changed: 1 addition & 1 deletion
diff --git a/‎flinkx-binlog/flinkx-binlog-reader/pom.xml‎
Lines changed: 0 additions & 4 deletions b/‎flinkx-binlog/flinkx-binlog-reader/pom.xml‎
Lines changed: 0 additions & 4 deletions
diff --git a/‎flinkx-clickhouse/flinkx-clickhouse-reader/src/main/java/com/dtstack/flinkx/clickhouse/format/ClickhouseInputFormat.java‎
Lines changed: 12 additions & 42 deletions b/‎flinkx-clickhouse/flinkx-clickhouse-reader/src/main/java/com/dtstack/flinkx/clickhouse/format/ClickhouseInputFormat.java‎
Lines changed: 12 additions & 42 deletions
diff --git a/‎flinkx-core/pom.xml‎
Lines changed: 7 additions & 7 deletions b/‎flinkx-core/pom.xml‎
Lines changed: 7 additions & 7 deletions
diff --git a/‎flinkx-core/src/main/java/com/dtstack/flinkx/Main.java‎
Lines changed: 1 addition & 4 deletions b/‎flinkx-core/src/main/java/com/dtstack/flinkx/Main.java‎
Lines changed: 1 addition & 4 deletions
diff --git a/‎flinkx-core/src/main/java/com/dtstack/flinkx/classloader/PluginUtil.java‎
Lines changed: 22 additions & 13 deletions b/‎flinkx-core/src/main/java/com/dtstack/flinkx/classloader/PluginUtil.java‎
Lines changed: 22 additions & 13 deletions
diff --git a/‎flinkx-core/src/main/java/com/dtstack/flinkx/config/ReaderConfig.java‎
Lines changed: 9 additions & 0 deletions b/‎flinkx-core/src/main/java/com/dtstack/flinkx/config/ReaderConfig.java‎
Lines changed: 9 additions & 0 deletions
@@ -19,13 +19,8 @@
         ......
       },
       "speed" : {
-      	......
-      },
-      "errorLimit" : {
-      	......
       },
       "dirty" : {
-      	......
       },
       "log" : {
       	......
@@ -53,12 +48,13 @@
 reader用于配置数据的输入源，即数据从何而来。具体配置如下所示：
 
 ```json
+{
 "reader" : {
   "name" : "xxreader",
   "parameter" : {
-		......
   }
 }
+}
 ```
 
 | 名称 | 说明 | 是否必填 |
 
@@ -69,6 +69,7 @@
 <br />1.读取全部字段，如果字段数量很多，可以使用下面的写法：
 ```json
 "column":["*"]
+{"column":["*"]}
 ```
 2.只指定字段名称：
 ```
 
@@ -466,4 +466,5 @@
       }
   }
 }
+}
 ```
@@ -18,7 +18,7 @@ cd flink-1.8.5
 ## 编译插件
 
 ```bash
-mvn clean package -Dmaven.test.skip=true
+mvn clean package -DskipTests
 ```
 
 ## 运行任务
 
@@ -25,10 +25,6 @@
                     <artifactId>logback-core</artifactId>
                     <groupId>ch.qos.logback</groupId>
                 </exclusion>
-                <exclusion>
-                    <artifactId>protobuf-java</artifactId>
-                    <groupId>com.google.protobuf</groupId>
-                </exclusion>
                 <exclusion>
                     <artifactId>jsr305</artifactId>
                     <groupId>com.google.code.findbugs</groupId>
 
@@ -20,14 +20,10 @@
 import com.dtstack.flinkx.clickhouse.core.ClickhouseUtil;
 import com.dtstack.flinkx.rdb.inputformat.JdbcInputFormat;
 import com.dtstack.flinkx.rdb.util.DbUtil;
-import com.dtstack.flinkx.reader.MetaColumn;
-import com.dtstack.flinkx.util.ClassUtil;
-import com.dtstack.flinkx.util.ExceptionUtil;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.flink.core.io.InputSplit;
 import org.apache.flink.types.Row;
 
 import java.io.IOException;
+import java.sql.Connection;
 import java.sql.SQLException;
 import java.util.ArrayList;
 import java.sql.Statement;
@@ -42,42 +38,6 @@
  */
 public class ClickhouseInputFormat extends JdbcInputFormat {
 
-    @Override
-    public void openInternal(InputSplit inputSplit) throws IOException {
-        try {
-            LOG.info("inputSplit = {}", inputSplit);
-            ClassUtil.forName(driverName, getClass().getClassLoader());
-            dbConn = ClickhouseUtil.getConnection(dbUrl, username, password);
-            statement = dbConn.createStatement(resultSetType, resultSetConcurrency);
-            statement.setFetchSize(fetchSize);
-            statement.setQueryTimeout(queryTimeOut);
-            String querySql = buildQuerySql(inputSplit);
-            resultSet = statement.executeQuery(querySql);
-            columnCount = resultSet.getMetaData().getColumnCount();
-
-            boolean splitWithRowCol = numPartitions > 1 && StringUtils.isNotEmpty(splitKey) && splitKey.contains("(");
-            if(splitWithRowCol){
-                columnCount = columnCount-1;
-            }
-            checkSize(columnCount, metaColumns);
-            hasNext = resultSet.next();
-
-            if (StringUtils.isEmpty(customSql)){
-                descColumnTypeList = DbUtil.analyzeTable(dbUrl, username, password, databaseInterface, table, metaColumns);
-            } else {
-                descColumnTypeList = new ArrayList<>();
-                for (MetaColumn metaColumn : metaColumns) {
-                    descColumnTypeList.add(metaColumn.getName());
-                }
-            }
-        } catch (Exception e) {
-            LOG.error("open failed,e = {}", ExceptionUtil.getErrorMessage(e));
-            throw new RuntimeException(e);
-        }
-
-        LOG.info("JdbcInputFormat[{}]open: end", jobName);
-    }
-
     @Override
     public Row nextRecordInternal(Row row) throws IOException {
         if (!hasNext) {
@@ -88,7 +48,7 @@ public Row nextRecordInternal(Row row) throws IOException {
             for (int pos = 0; pos < row.getArity(); pos++) {
                 Object obj = resultSet.getObject(pos + 1);
                 if(obj != null) {
-                    obj = clobToString(obj);
+                    obj = DbUtil.clobToString(obj);
                 }
                 row.setField(pos, obj);
             }
@@ -99,4 +59,14 @@ public Row nextRecordInternal(Row row) throws IOException {
             throw new IOException("Couldn't access resultSet", npe);
         }
     }
+
+    /**
+     * 获取数据库连接，用于子类覆盖
+     * @return connection
+     * @throws SQLException
+     */
+    @Override
+    protected Connection getConnection() throws SQLException {
+        return ClickhouseUtil.getConnection(dbUrl, username, password);
+    }
 }
@@ -86,6 +86,12 @@
             </exclusions>
         </dependency>
 
+        <dependency>
+            <groupId>commons-cli</groupId>
+            <artifactId>commons-cli</artifactId>
+            <version>1.2</version>
+        </dependency>
+
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-yarn_2.11</artifactId>
@@ -117,7 +123,7 @@
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-databind</artifactId>
-            <version>2.9.10.3</version>
+            <version>2.9.10.1</version>
         </dependency>
         <dependency>
             <groupId>io.prometheus</groupId>
@@ -129,11 +135,6 @@
             <artifactId>simpleclient</artifactId>
             <version>0.5.0</version>
         </dependency>
-        <dependency>
-            <groupId>org.apache.avro</groupId>
-            <artifactId>avro</artifactId>
-            <version>1.8.2</version>
-        </dependency>
     </dependencies>
 
     <build>
@@ -173,7 +174,6 @@
                                     <include>org.apache.httpcomponents:*</include>
                                     <include>io.prometheus:*</include>
                                     <include>org.apache.avro:*</include>
-                                    <include>org.codehaus.jackson:*</include>
                                     <include>com.fasterxml.jackson.core:*</include>
                                     <include>commons-*:*</include>
                                 </includes>
 
@@ -46,17 +46,14 @@
 import org.apache.flink.streaming.api.datastream.DataStreamSink;
 import org.apache.flink.streaming.api.datastream.DataStreamSource;
 import org.apache.flink.streaming.api.environment.CheckpointConfig;
-import org.apache.flink.streaming.api.environment.StreamContextEnvironment;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.types.Row;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.lang.reflect.Field;
 import java.net.URL;
 import java.net.URLDecoder;
 import java.util.ArrayList;
-import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 import java.util.Set;
@@ -87,8 +84,8 @@ public static void main(String[] args) throws Exception {
         String jobIdString = options.getJobid();
         String monitor = options.getMonitor();
         String pluginRoot = options.getPluginRoot();
-        String remotePluginPath = options.getRemotePluginPath();
         String savepointPath = options.getS();
+        String remotePluginPath = options.getRemotePluginPath();
         Properties confProperties = parseConf(options.getConfProp());
 
         // 解析jobPath指定的任务配置文件
 
@@ -20,11 +20,15 @@
 package com.dtstack.flinkx.classloader;
 
 import com.dtstack.flinkx.util.SysUtil;
+import org.apache.commons.lang3.StringUtils;
 
 import java.io.File;
+import java.io.FilenameFilter;
 import java.net.MalformedURLException;
 import java.net.URL;
+import java.util.ArrayList;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Set;
 
 /**
@@ -41,22 +45,27 @@ public class PluginUtil {
 
     private static final String PACKAGE_PREFIX = "com.dtstack.flinkx.";
 
+    private static final String JAR_PREFIX = "flinkx";
+
+    private static final String SP = File.separator;
+
     public static Set<URL> getJarFileDirPath(String pluginName, String pluginRoot, String remotePluginPath) {
         Set<URL> urlList = new HashSet<>();
+        List<File> pathDir = new ArrayList<>();
 
-        File commonDir = new File(pluginRoot + File.separator + COMMON_DIR + File.separator);
-        File pluginDir = new File(pluginRoot + File.separator + pluginName);
-        File remoteDir = null;
+        if (pluginRoot != null) {
+            pathDir.add(new File(pluginRoot + File.separator + pluginName));
+            pathDir.add(new File(pluginRoot + File.separator + COMMON_DIR + File.separator));
+        }
 
         if (remotePluginPath != null) {
-            remoteDir = new File(remotePluginPath + File.separator + pluginName);
+            pathDir.add(new File(remotePluginPath + File.separator + pluginName));
+            pathDir.add(new File(remotePluginPath + File.separator + COMMON_DIR + File.separator));
         }
 
         try {
-            urlList.addAll(SysUtil.findJarsInDir(commonDir));
-            urlList.addAll(SysUtil.findJarsInDir(pluginDir));
-            if(remoteDir!=null){
-                urlList.addAll(SysUtil.findJarsInDir(remoteDir));
+            for(File path : pathDir) {
+                urlList.addAll(SysUtil.findJarsInDir(path));
             }
 
             return urlList;
@@ -65,11 +74,11 @@ public static Set<URL> getJarFileDirPath(String pluginName, String pluginRoot, S
         }
     }
 
-    public static String getPluginClassName(String pluginName){
+    public static String getPluginClassName(String pluginName) {
         String pluginClassName;
-        if(pluginName.toLowerCase().endsWith(READER_SUFFIX)) {
+        if (pluginName.toLowerCase().endsWith(READER_SUFFIX)) {
             pluginClassName = PACKAGE_PREFIX + camelize(pluginName, READER_SUFFIX);
-        } else if(pluginName.toLowerCase().endsWith(WRITER_SUFFIX)) {
+        } else if (pluginName.toLowerCase().endsWith(WRITER_SUFFIX)) {
             pluginClassName = PACKAGE_PREFIX + camelize(pluginName, WRITER_SUFFIX);
         } else {
             throw new IllegalArgumentException("Plugin Name should end with reader, writer or database");
@@ -85,8 +94,8 @@ private static String camelize(String pluginName, String suffix) {
         suffix = suffix.toLowerCase();
         StringBuffer sb = new StringBuffer();
         sb.append(left + "." + suffix + ".");
-        sb.append(left.substring(0,1).toUpperCase() + left.substring(1));
-        sb.append(suffix.substring(0,1).toUpperCase() + suffix.substring(1));
+        sb.append(left.substring(0, 1).toUpperCase() + left.substring(1));
+        sb.append(suffix.substring(0, 1).toUpperCase() + suffix.substring(1));
         return sb.toString();
     }
 }
@@ -95,6 +95,7 @@ public void setColumn(List column) {
         public class ConnectionConfig extends AbstractConfig {
 
             public static final String KEY_TABLE_LIST = "table";
+            public static final String KEY_SCHEMA = "schema";
             public static final String KEY_JDBC_URL_LIST = "jdbcUrl";
             public static final String KEY_JDBC_USERNAME = "username";
             public static final String KEY_JDBC_PASSWORD = "password";
@@ -111,6 +112,14 @@ public void setTable(List<String> table) {
                 setVal(KEY_TABLE_LIST, table);
             }
 
+            public String getSchema(){
+                return (String) getVal(KEY_SCHEMA);
+            }
+
+            public void setSchema(String schema){
+                setVal(KEY_SCHEMA, schema);
+            }
+
             public List<String> getJdbcUrl() {
                 return (List<String>) getVal(KEY_JDBC_URL_LIST);
             }
Original file line number	Diff line number	Diff line change
`@@ -466,4 +466,5 @@`
`466`	`466`	`}`
`467`	`467`	`}`
`468`	`468`	`}`
	`469`	`+}`
`469`	`470`	```