Merge pull request #22 from taosdata/test/TS-1218

[TS-1219]<feature>: incremental synchronization scripts of datax

tdenginereader/doc/tdenginereader-CN.md

@@ -37,8 +37,7 @@ TDengineReader 通过TDengine的JDBC driver查询获取数据。
               "phase"
             ],
             "beginDateTime": "2017-07-14 10:40:00",
-            "endDateTime": "2017-08-14 10:40:00",
-            "splitInterval": "1d"
+            "endDateTime": "2017-08-14 10:40:00"
           }
         },
         "writer": {
@@ -79,17 +78,12 @@ TDengineReader 通过TDengine的JDBC driver查询获取数据。
     * 必选：是 <br />
     * 默认值：无<br />
 * **beginDateTime**
-    * 描述：数据的开始时间，Job迁移从begineDateTime到endDateTime的数据，格式为yyyy-MM-dd HH:mm:ss <br />
-    * 必选：是 <br />
+    * 描述：数据的开始时间，Job迁移从begineDateTime到endDateTime的数据，格式为yyyy-MM-dd HH:mm:ss，如果不填为全量同步 <br />
+    * 必选：否 <br />
     * 默认值：无 <br />
 * **endDateTime**
-    * 描述：数据的结束时间，Job迁移从begineDateTime到endDateTime的数据，格式为yyyy-MM-dd HH:mm:ss <br />
-    * 必选：是 <br />
-    * 默认值：无 <br />
-* **splitInterval**
-    * 描述：按照splitInterval来划分task, 每splitInterval创建一个task。例如，20d代表按照每20天的数据划分为1个task。
-      可以配置的时间单位为：d（天）, h（小时）, m（分钟）, s（秒） <br />
-    * 必选：是 <br />
+    * 描述：数据的结束时间，Job迁移从begineDateTime到endDateTime的数据，格式为yyyy-MM-dd HH:mm:ss，如果不填为全量同步 <br />
+    * 必选：否 <br />
     * 默认值：无 <br />
 
 ### 3.3 类型转换
@@ -106,7 +100,6 @@ TDengineReader 通过TDengine的JDBC driver查询获取数据。
 | BINARY          | Bytes         |
 | NCHAR           | String        |
 
-
 ## 4 性能报告
 
 ### 4.1 环境准备

tdenginereader/pom.xml

@@ -29,10 +29,29 @@
             </exclusions>
         </dependency>
 
+        <dependency>
+            <groupId>com.alibaba</groupId>
+            <artifactId>fastjson</artifactId>
+            <version>1.2.78</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.alibaba.datax.tdenginewriter</groupId>
+            <artifactId>tdenginewriter</artifactId>
+            <version>0.0.1-SNAPSHOT</version>
+            <scope>compile</scope>
+        </dependency>
+
         <dependency>
             <groupId>com.taosdata.jdbc</groupId>
             <artifactId>taos-jdbcdriver</artifactId>
             <version>2.0.37</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>com.alibaba</groupId>
+                    <artifactId>fastjson</artifactId>
+                </exclusion>
+            </exclusions>
         </dependency>
 
         <dependency>
@@ -47,6 +66,21 @@
             <version>0.0.1-SNAPSHOT</version>
             <scope>compile</scope>
         </dependency>
+
+        <dependency>
+            <groupId>com.alibaba.datax</groupId>
+            <artifactId>datax-core</artifactId>
+            <version>0.0.1-SNAPSHOT</version>
+            <scope>test</scope>
+        </dependency>
+        <!-- 添加 dm8 jdbc jar 包依赖-->
+<!--        <dependency>-->
+<!--            <groupId>com.dameng</groupId>-->
+<!--            <artifactId>dm-jdbc</artifactId>-->
+<!--            <version>1.8</version>-->
+<!--            <scope>system</scope>-->
+<!--            <systemPath>${project.basedir}/src/test/resources/DmJdbcDriver18.jar</systemPath>-->
+<!--        </dependency>-->
     </dependencies>
 
     <build>

tdenginereader/src/main/java/com/alibaba/datax/plugin/reader/Key.java

@@ -1,19 +0,0 @@
-package com.alibaba.datax.plugin.reader;
-
-public class Key {
-
-    public static final String JDBC_URL = "jdbcUrl";
-//    public static final String HOST = "host";
-//    public static final String PORT = "port";
-//    public static final String DB = "db";
-    public static final String TABLE = "table";
-    public static final String USER = "username";
-    public static final String PASSWORD = "password";
-    public static final String CONNECTION = "connection";
-//    public static final String SQL = "sql";
-    public static final String BEGIN_DATETIME = "beginDateTime";
-    public static final String END_DATETIME = "endDateTime";
-    public static final String SPLIT_INTERVAL = "splitInterval";
-    public static final String COLUMN = "column";
-    public static final String MANDATORY_ENCODING = "mandatoryEncoding";
-}

tdenginereader/src/main/java/com/alibaba/datax/plugin/reader/TDengineReader.java

@@ -1,13 +1,11 @@
 package com.alibaba.datax.plugin.reader;
 
-import com.alibaba.datax.common.constant.CommonConstant;
 import com.alibaba.datax.common.element.*;
 import com.alibaba.datax.common.exception.DataXException;
 import com.alibaba.datax.common.plugin.RecordSender;
 import com.alibaba.datax.common.spi.Reader;
 import com.alibaba.datax.common.util.Configuration;
-import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
-import com.alibaba.fastjson.JSON;
+import com.alibaba.datax.plugin.writer.tdenginewriter.Key;
 import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -22,84 +20,77 @@ import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
 public class TDengineReader extends Reader {
+
     private static final String DATETIME_FORMAT = "yyyy-MM-dd HH:mm:ss";
 
     public static class Job extends Reader.Job {
-
         private static final Logger LOG = LoggerFactory.getLogger(Job.class);
         private Configuration originalConfig;
 
         @Override
         public void init() {
             this.originalConfig = super.getPluginJobConf();
-            // check user
-            String user = this.originalConfig.getString(Key.USER);
-            if (StringUtils.isBlank(user))
-                throw DataXException.asDataXException(TDengineReaderErrorCode.REQUIRED_VALUE, "The parameter [" + Key.USER + "] is not set.");
+            // check username
+            String username = this.originalConfig.getString(Key.USERNAME);
+            if (StringUtils.isBlank(username))
+                throw DataXException.asDataXException(TDengineReaderErrorCode.REQUIRED_VALUE,
+                        "The parameter [" + Key.USERNAME + "] is not set.");
 
             // check password
             String password = this.originalConfig.getString(Key.PASSWORD);
             if (StringUtils.isBlank(password))
-                throw DataXException.asDataXException(TDengineReaderErrorCode.REQUIRED_VALUE, "The parameter [" + Key.PASSWORD + "] is not set.");
-
-            SimpleDateFormat format = new SimpleDateFormat(DATETIME_FORMAT);
-            // check beginDateTime
-            String beginDatetime = this.originalConfig.getString(Key.BEGIN_DATETIME);
-            if (StringUtils.isBlank(beginDatetime))
-                throw DataXException.asDataXException(TDengineReaderErrorCode.REQUIRED_VALUE, "The parameter [" + Key.BEGIN_DATETIME + "] is not set.");
-            Long start;
-            try {
-                start = format.parse(beginDatetime).getTime();
-            } catch (ParseException e) {
-                throw DataXException.asDataXException(TDengineReaderErrorCode.ILLEGAL_VALUE, "The parameter [" + Key.BEGIN_DATETIME + "] needs to conform to the [" + DATETIME_FORMAT + "] format.");
-            }
-
-            // check endDateTime
-            String endDatetime = this.originalConfig.getString(Key.END_DATETIME);
-            if (StringUtils.isBlank(endDatetime))
-                throw DataXException.asDataXException(TDengineReaderErrorCode.REQUIRED_VALUE, "The parameter [" + Key.END_DATETIME + "] is not set.");
-            Long end;
-            try {
-                end = format.parse(endDatetime).getTime();
-            } catch (ParseException e) {
-                throw DataXException.asDataXException(TDengineReaderErrorCode.ILLEGAL_VALUE, "The parameter [" + Key.END_DATETIME + "] needs to conform to the [" + DATETIME_FORMAT + "] format.");
-            }
-            if (start >= end)
-                throw DataXException.asDataXException(TDengineReaderErrorCode.ILLEGAL_VALUE, "The parameter [" + Key.BEGIN_DATETIME + "] should be less than the parameter [" + Key.END_DATETIME + "].");
-
-            // check splitInterval
-            String splitInterval = this.originalConfig.getString(Key.SPLIT_INTERVAL);
-            Long split;
-            if (StringUtils.isBlank(splitInterval))
-                throw DataXException.asDataXException(TDengineReaderErrorCode.REQUIRED_VALUE, "The parameter [" + Key.SPLIT_INTERVAL + "] is not set.");
-            try {
-                split = parseSplitInterval(splitInterval);
-            } catch (Exception e) {
-                throw DataXException.asDataXException(TDengineReaderErrorCode.ILLEGAL_VALUE, "The parameter [" + Key.SPLIT_INTERVAL + "] should be like: \"123d|h|m|s\", error: " + e.getMessage());
-            }
-
-            this.originalConfig.set(Key.BEGIN_DATETIME, start);
-            this.originalConfig.set(Key.END_DATETIME, end);
-            this.originalConfig.set(Key.SPLIT_INTERVAL, split);
+                throw DataXException.asDataXException(TDengineReaderErrorCode.REQUIRED_VALUE,
+                        "The parameter [" + Key.PASSWORD + "] is not set.");
 
             // check connection
             List<Object> connection = this.originalConfig.getList(Key.CONNECTION);
             if (connection == null || connection.isEmpty())
-                throw DataXException.asDataXException(TDengineReaderErrorCode.REQUIRED_VALUE, "The parameter [" + Key.CONNECTION + "] is not set.");
+                throw DataXException.asDataXException(TDengineReaderErrorCode.REQUIRED_VALUE,
+                        "The parameter [" + Key.CONNECTION + "] is not set.");
             for (int i = 0; i < connection.size(); i++) {
                 Configuration conn = Configuration.from(connection.get(i).toString());
                 List<Object> table = conn.getList(Key.TABLE);
                 if (table == null || table.isEmpty())
-                    throw DataXException.asDataXException(TDengineReaderErrorCode.REQUIRED_VALUE, "The parameter [" + Key.TABLE + "] of connection[" + (i + 1) + "] is not set.");
+                    throw DataXException.asDataXException(TDengineReaderErrorCode.REQUIRED_VALUE,
+                            "The parameter [" + Key.TABLE + "] of connection[" + (i + 1) + "] is not set.");
                 String jdbcUrl = conn.getString(Key.JDBC_URL);
                 if (StringUtils.isBlank(jdbcUrl))
-                    throw DataXException.asDataXException(TDengineReaderErrorCode.REQUIRED_VALUE, "The parameter [" + Key.JDBC_URL + "] of connection[" + (i + 1) + "] is not set.");
+                    throw DataXException.asDataXException(TDengineReaderErrorCode.REQUIRED_VALUE,
+                            "The parameter [" + Key.JDBC_URL + "] of connection[" + (i + 1) + "] is not set.");
             }
-
             // check column
             List<Object> column = this.originalConfig.getList(Key.COLUMN);
             if (column == null || column.isEmpty())
-                throw DataXException.asDataXException(TDengineReaderErrorCode.REQUIRED_VALUE, "The parameter [" + Key.CONNECTION + "] is not set or is empty.");
+                throw DataXException.asDataXException(TDengineReaderErrorCode.REQUIRED_VALUE,
+                        "The parameter [" + Key.CONNECTION + "] is not set or is empty.");
+
+            SimpleDateFormat format = new SimpleDateFormat(DATETIME_FORMAT);
+            // check beginDateTime
+            String beginDatetime = this.originalConfig.getString(Key.BEGIN_DATETIME);
+            long start = Long.MIN_VALUE;
+            if (!StringUtils.isBlank(beginDatetime)) {
+                try {
+                    start = format.parse(beginDatetime).getTime();
+                } catch (ParseException e) {
+                    throw DataXException.asDataXException(TDengineReaderErrorCode.ILLEGAL_VALUE,
+                            "The parameter [" + Key.BEGIN_DATETIME + "] needs to conform to the [" + DATETIME_FORMAT + "] format.");
+                }
+            }
+            // check endDateTime
+            String endDatetime = this.originalConfig.getString(Key.END_DATETIME);
+            long end = Long.MAX_VALUE;
+            if (!StringUtils.isBlank(endDatetime)) {
+                try {
+                    end = format.parse(endDatetime).getTime();
+                } catch (ParseException e) {
+                    throw DataXException.asDataXException(TDengineReaderErrorCode.ILLEGAL_VALUE,
+                            "The parameter [" + Key.END_DATETIME + "] needs to conform to the [" + DATETIME_FORMAT + "] format.");
+                }
+            }
+            if (start >= end)
+                throw DataXException.asDataXException(TDengineReaderErrorCode.ILLEGAL_VALUE,
+                        "The parameter [" + Key.BEGIN_DATETIME + "] should be less than the parameter [" + Key.END_DATETIME + "].");
+
         }
 
         @Override
@@ -110,78 +101,58 @@ public class TDengineReader extends Reader {
         @Override
         public List<Configuration> split(int adviceNumber) {
             List<Configuration> configurations = new ArrayList<>();
-            // do split
-            Long start = this.originalConfig.getLong(Key.BEGIN_DATETIME);
-            Long end = this.originalConfig.getLong(Key.END_DATETIME);
-            Long split = this.originalConfig.getLong(Key.SPLIT_INTERVAL);
 
-            List<Object> conns = this.originalConfig.getList(Key.CONNECTION);
-
-            for (Long ts = start; ts < end; ts += split) {
-                for (int i = 0; i < conns.size(); i++) {
-                    Configuration clone = this.originalConfig.clone();
-                    clone.remove(Key.SPLIT_INTERVAL);
-
-                    clone.set(Key.BEGIN_DATETIME, ts);
-                    clone.set(Key.END_DATETIME, Math.min(ts + split, end));
-
-                    Configuration conf = Configuration.from(conns.get(i).toString());
-                    String jdbcUrl = conf.getString(Key.JDBC_URL);
-                    clone.set(Key.JDBC_URL, jdbcUrl);
-                    clone.set(Key.TABLE, conf.getList(Key.TABLE));
-
-                    // 抽取 jdbcUrl 中的 ip/port 进行资源使用的打标，以提供给 core 做有意义的 shuffle 操作
-                    clone.set(CommonConstant.LOAD_BALANCE_RESOURCE_MARK, DataBaseType.parseIpFromJdbcUrl(jdbcUrl));
-                    clone.remove(Key.CONNECTION);
-
-                    configurations.add(clone);
-                    LOG.info("Configuration: {}", JSON.toJSONString(clone));
-                }
+            List<Object> connectionList = this.originalConfig.getList(Key.CONNECTION);
+            for (Object conn : connectionList) {
+                Configuration clone = this.originalConfig.clone();
+                Configuration conf = Configuration.from(conn.toString());
+                String jdbcUrl = conf.getString(Key.JDBC_URL);
+                clone.set(Key.JDBC_URL, jdbcUrl);
+                clone.set(Key.TABLE, conf.getList(Key.TABLE));
+                clone.remove(Key.CONNECTION);
+                configurations.add(clone);
             }
+
+            LOG.info("Configuration: {}", configurations);
             return configurations;
         }
     }
 
     public static class Task extends Reader.Task {
         private static final Logger LOG = LoggerFactory.getLogger(Task.class);
-        private Connection conn;
 
-        private Long startTime;
-        private Long endTime;
+        private Configuration readerSliceConfig;
+        private Connection conn;
         private List<String> tables;
         private List<String> columns;
-        private String mandatoryEncoding;
+
+        private String startTime;
+        private String endTime;
 
         @Override
         public void init() {
-            Configuration readerSliceConfig = super.getPluginJobConf();
-            LOG.info("getPluginJobConf: {}", JSON.toJSONString(readerSliceConfig));
+            this.readerSliceConfig = super.getPluginJobConf();
+            LOG.info("getPluginJobConf: {}", readerSliceConfig);
 
             String url = readerSliceConfig.getString(Key.JDBC_URL);
             if (StringUtils.isBlank(url))
                 throw DataXException.asDataXException(TDengineReaderErrorCode.REQUIRED_VALUE,
                         "The parameter [" + Key.JDBC_URL + "] is not set.");
-
-            tables = readerSliceConfig.getList(Key.TABLE, String.class);
-            columns = readerSliceConfig.getList(Key.COLUMN, String.class);
-
-            String user = readerSliceConfig.getString(Key.USER);
+            String user = readerSliceConfig.getString(Key.USERNAME);
             String password = readerSliceConfig.getString(Key.PASSWORD);
-
             try {
-                conn = DriverManager.getConnection(url, user, password);
+                this.conn = DriverManager.getConnection(url, user, password);
             } catch (SQLException e) {
                 throw DataXException.asDataXException(TDengineReaderErrorCode.CONNECTION_FAILED,
-                        "The parameter [" + Key.JDBC_URL + "] : " + url + " failed to connect since: " + e.getMessage());
+                        "The parameter [" + Key.JDBC_URL + "] : " + url + " failed to connect since: " + e.getMessage(), e);
             }
 
-            this.mandatoryEncoding = readerSliceConfig.getString(Key.MANDATORY_ENCODING, "");
-
-            startTime = readerSliceConfig.getLong(Key.BEGIN_DATETIME);
-            endTime = readerSliceConfig.getLong(Key.END_DATETIME);
+            this.tables = readerSliceConfig.getList(Key.TABLE, String.class);
+            this.columns = readerSliceConfig.getList(Key.COLUMN, String.class);
+            this.startTime = readerSliceConfig.getString(Key.BEGIN_DATETIME);
+            this.endTime = readerSliceConfig.getString(Key.END_DATETIME);
         }
 
-
         @Override
         public void destroy() {
 
@@ -189,18 +160,31 @@ public class TDengineReader extends Reader {
 
         @Override
         public void startRead(RecordSender recordSender) {
+
             try (Statement stmt = conn.createStatement()) {
                 for (String table : tables) {
-                    String sql = "select " + StringUtils.join(columns, ",") + " from " + table
-                            + " where _c0 >= " + startTime + " and _c0 < " + endTime;
+                    StringBuilder sb = new StringBuilder();
+                    sb.append("select ").append(StringUtils.join(columns, ","))
+                            .append(" from ").append(table).append(" ");
+
+                    if (StringUtils.isBlank(startTime)) {
+                        sb.append("where _c0 >= ").append(Long.MIN_VALUE);
+                    } else {
+                        sb.append("where _c0 >= '").append(startTime).append("'");
+                    }
+                    if (!StringUtils.isBlank(endTime)) {
+                        sb.append(" and _c0 < '").append(endTime).append("'");
+                    }
+
+                    String sql = sb.toString();
                     ResultSet rs = stmt.executeQuery(sql);
-                    ResultSetMetaData metaData = rs.getMetaData();
                     while (rs.next()) {
-                        transportOneRecord(recordSender, rs, metaData, metaData.getColumnCount(), this.mandatoryEncoding);
+                        Record record = buildRecord(recordSender, rs, "UTF-8");
+                        recordSender.sendToWriter(record);
                     }
                 }
             } catch (SQLException e) {
-                throw DataXException.asDataXException(TDengineReaderErrorCode.ILLEGAL_VALUE, "获取或发送数据点的过程中出错！", e);
+                throw DataXException.asDataXException(TDengineReaderErrorCode.RUNTIME_EXCEPTION, e.getMessage(), e);
             } finally {
                 try {
                     if (conn != null)
@@ -211,17 +195,11 @@ public class TDengineReader extends Reader {
             }
         }
 
-        private Record transportOneRecord(RecordSender recordSender, ResultSet rs, ResultSetMetaData metaData, int columnCount, String mandatoryEncoding) {
-            Record record = buildRecord(recordSender, rs, metaData, columnCount, mandatoryEncoding);
-            recordSender.sendToWriter(record);
-            return record;
-        }
-
-        private Record buildRecord(RecordSender recordSender, ResultSet rs, ResultSetMetaData metaData, int columnCount, String mandatoryEncoding) {
+        private Record buildRecord(RecordSender recordSender, ResultSet rs, String mandatoryEncoding) {
             Record record = recordSender.createRecord();
-
             try {
-                for (int i = 1; i <= columnCount; i++) {
+                ResultSetMetaData metaData = rs.getMetaData();
+                for (int i = 1; i <= metaData.getColumnCount(); i++) {
                     int columnType = metaData.getColumnType(i);
                     switch (columnType) {
                         case Types.SMALLINT:
@@ -261,16 +239,16 @@ public class TDengineReader extends Reader {
         }
     }
 
-    private static final long second = 1000;
-    private static final long minute = 60 * second;
-    private static final long hour = 60 * minute;
-    private static final long day = 24 * hour;
-
     private static Long parseSplitInterval(String splitInterval) throws Exception {
+        final long second = 1000;
+        final long minute = 60 * second;
+        final long hour = 60 * minute;
+        final long day = 24 * hour;
+
         Pattern compile = Pattern.compile("^(\\d+)([dhms])$");
         Matcher matcher = compile.matcher(splitInterval);
         while (matcher.find()) {
-            Long value = Long.valueOf(matcher.group(1));
+            long value = Long.parseLong(matcher.group(1));
             if (value == 0)
                 throw new Exception("invalid splitInterval: 0");
             char unit = matcher.group(2).charAt(0);

tdenginereader/src/main/java/com/alibaba/datax/plugin/reader/TDengineReaderErrorCode.java

@@ -4,9 +4,10 @@ import com.alibaba.datax.common.spi.ErrorCode;
 
 public enum TDengineReaderErrorCode implements ErrorCode {
 
-    REQUIRED_VALUE("TDengineReader-00", "缺失必要的值"),
-    ILLEGAL_VALUE("TDengineReader-01", "值非法"),
-    CONNECTION_FAILED("TDengineReader-02", "连接错误");
+    REQUIRED_VALUE("TDengineReader-00", "parameter value is missing"),
+    ILLEGAL_VALUE("TDengineReader-01", "invalid parameter value"),
+    CONNECTION_FAILED("TDengineReader-02", "connection error"),
+    RUNTIME_EXCEPTION("TDengineWriter-03", "runtime exception");
 
     private final String code;
     private final String description;

tdenginereader/src/test/java/com/alibaba/datax/plugin/reader/TDengine2DMTest.java

@@ -0,0 +1,87 @@
+package com.alibaba.datax.plugin.reader;
+
+import com.alibaba.datax.core.Engine;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Random;
+
+@Ignore
+public class TDengine2DMTest {
+    private static final String host1 = "192.168.56.105";
+    private static final String host2 = "192.168.0.72";
+
+    private final Random random = new Random(System.currentTimeMillis());
+
+    @Test
+    public void t2dm_case01() throws Throwable {
+        // given
+        createSupTable("ms");
+
+        // when
+        String[] params = {"-mode", "standalone", "-jobid", "-1", "-job", "src/test/resources/t2dm.json"};
+        System.setProperty("datax.home", "../target/datax/datax");
+        Engine.entry(params);
+    }
+
+    @Test
+    public void t2dm_case02() throws Throwable {
+        // given
+        createSupTable("us");
+
+        // when
+        String[] params = {"-mode", "standalone", "-jobid", "-1", "-job", "src/test/resources/t2dm.json"};
+        System.setProperty("datax.home", "../target/datax/datax");
+        Engine.entry(params);
+    }
+
+    @Test
+    public void t2dm_case03() throws Throwable {
+        // given
+        createSupTable("ns");
+
+        // when
+        String[] params = {"-mode", "standalone", "-jobid", "-1", "-job", "src/test/resources/t2dm.json"};
+        System.setProperty("datax.home", "../target/datax/datax");
+        Engine.entry(params);
+    }
+
+    private void createSupTable(String precision) throws SQLException {
+        final String url = "jdbc:TAOS-RS://" + host1 + ":6041/";
+        try (Connection conn = DriverManager.getConnection(url, "root", "taosdata")) {
+            Statement stmt = conn.createStatement();
+
+            stmt.execute("drop database if exists db1");
+            stmt.execute("create database if not exists db1 precision '" + precision + "'");
+            stmt.execute("create table db1.stb1(ts timestamp, f1 tinyint, f2 smallint, f3 int, f4 bigint, f5 float, " +
+                    "f6 double, f7 bool, f8 binary(100), f9 nchar(100)) tags(t1 timestamp, t2 tinyint, t3 smallint, " +
+                    "t4 int, t5 bigint, t6 float, t7 double, t8 bool, t9 binary(100), t10 nchar(100))");
+
+            for (int i = 1; i <= 10; i++) {
+                stmt.execute("insert into db1.tb" + i + " using db1.stb1 tags(now, " + random.nextInt(10) + "," +
+                        random.nextInt(10) + "," + random.nextInt(10) + "," + random.nextInt(10) + "," +
+                        random.nextFloat() + "," + random.nextDouble() + "," + random.nextBoolean() + ",'abcABC123'," +
+                        "'北京朝阳望京') values(now+" + i + "s, " + random.nextInt(10) + "," + random.nextInt(10) + "," +
+                        +random.nextInt(10) + "," + random.nextInt(10) + "," + random.nextFloat() + "," +
+                        random.nextDouble() + "," + random.nextBoolean() + ",'abcABC123','北京朝阳望京')");
+            }
+            stmt.close();
+        }
+
+        final String url2 = "jdbc:dm://" + host2 + ":5236";
+        try (Connection conn = DriverManager.getConnection(url2, "TESTUSER", "test123456")) {
+            conn.setAutoCommit(true);
+            Statement stmt = conn.createStatement();
+            stmt.execute("drop table if exists stb2");
+            stmt.execute("create table stb2(ts timestamp, f1 tinyint, f2 smallint, f3 int, f4 bigint, f5 float, " +
+                    "f6 double, f7 BIT, f8 VARCHAR(100), f9 VARCHAR2(200), t1 timestamp, t2 tinyint, t3 smallint, " +
+                    "t4 int, t5 bigint, t6 float, t7 double, t8 BIT, t9 VARCHAR(100), t10 VARCHAR2(200))");
+        }
+    }
+
+}

tdenginereader/src/test/java/com/alibaba/datax/plugin/reader/TDengineReaderTest.java

@@ -1,72 +1,118 @@
 package com.alibaba.datax.plugin.reader;
 
 import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.writer.tdenginewriter.Key;
 import org.junit.Assert;
-import org.junit.Before;
 import org.junit.Test;
 
-import java.text.ParseException;
-import java.text.SimpleDateFormat;
 import java.util.List;
 
 public class TDengineReaderTest {
 
-    TDengineReader.Job job;
-
-    @Before
-    public void before() {
-        job = new TDengineReader.Job();
+    @Test
+    public void jobInit_case01() {
+        // given
+        TDengineReader.Job job = new TDengineReader.Job();
         Configuration configuration = Configuration.from("{" +
-                "\"user\": \"root\"," +
+                "\"username\": \"root\"," +
                 "\"password\": \"taosdata\"," +
                 "\"connection\": [{\"table\":[\"weather\"],\"jdbcUrl\":\"jdbc:TAOS-RS://master:6041/test\"}]," +
                 "\"column\": [\"ts\",\"current\",\"voltage\",\"phase\"]," +
                 "\"beginDateTime\": \"2021-01-01 00:00:00\"," +
-                "\"endDateTime\": \"2021-01-01 12:00:00\"," +
-                "\"splitInterval\": \"1h\"" +
+                "\"endDateTime\": \"2021-01-01 12:00:00\"" +
                 "}");
         job.setPluginJobConf(configuration);
-    }
 
-    @Test
-    public void jobInit() throws ParseException {
         // when
         job.init();
 
         // assert
         Configuration conf = job.getPluginJobConf();
 
-        Assert.assertEquals("root", conf.getString("user"));
+        Assert.assertEquals("root", conf.getString(Key.USERNAME));
         Assert.assertEquals("taosdata", conf.getString("password"));
         Assert.assertEquals("weather", conf.getString("connection[0].table[0]"));
         Assert.assertEquals("jdbc:TAOS-RS://master:6041/test", conf.getString("connection[0].jdbcUrl"));
+        Assert.assertEquals("2021-01-01 00:00:00", conf.getString("beginDateTime"));
+        Assert.assertEquals("2021-01-01 12:00:00", conf.getString("endDateTime"));
+    }
 
-        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
 
-        Long start = sdf.parse("2021-01-01 00:00:00").getTime();
-        Assert.assertEquals(start, conf.getLong("beginDateTime"));
+    @Test
+    public void jobInit_case02() {
+        // given
+        TDengineReader.Job job = new TDengineReader.Job();
+        Configuration configuration = Configuration.from("{" +
+                "\"username\": \"root\"," +
+                "\"password\": \"taosdata\"," +
+                "\"connection\": [{\"table\":[\"weather\"],\"jdbcUrl\":\"jdbc:TAOS-RS://master:6041/test\"}]," +
+                "\"column\": [\"ts\",\"current\",\"voltage\",\"phase\"]" +
+                "}");
+        job.setPluginJobConf(configuration);
 
-        Long end = sdf.parse("2021-01-01 12:00:00").getTime();
-        Assert.assertEquals(end, conf.getLong("endDateTime"));
+        // when
+        job.init();
 
-        Assert.assertEquals(new Long(3600 * 1000), conf.getLong("splitInterval"));
+        // assert
+        Configuration conf = job.getPluginJobConf();
+
+        Assert.assertEquals("root", conf.getString(Key.USERNAME));
+        Assert.assertEquals("taosdata", conf.getString("password"));
+        Assert.assertEquals("weather", conf.getString("connection[0].table[0]"));
+        Assert.assertEquals("jdbc:TAOS-RS://master:6041/test", conf.getString("connection[0].jdbcUrl"));
     }
 
     @Test
-    public void jobSplit() {
+    public void jobSplit_case01() {
+        // given
+        TDengineReader.Job job = new TDengineReader.Job();
+        Configuration configuration = Configuration.from("{" +
+                "\"username\": \"root\"," +
+                "\"password\": \"taosdata\"," +
+                "\"connection\": [{\"table\":[\"weather\"],\"jdbcUrl\":\"jdbc:TAOS-RS://master:6041/test\"}]," +
+                "\"column\": [\"ts\",\"current\",\"voltage\",\"phase\"]," +
+                "\"beginDateTime\": \"2021-01-01 00:00:00\"," +
+                "\"endDateTime\": \"2021-01-01 12:00:00\"" +
+                "}");
+        job.setPluginJobConf(configuration);
+
         // when
         job.init();
         List<Configuration> configurationList = job.split(1);
 
         // assert
-        Assert.assertEquals(12, configurationList.size());
-        for (int i = 0; i < configurationList.size(); i++) {
-            Configuration conf = configurationList.get(i);
-            Assert.assertEquals("root", conf.getString("user"));
-            Assert.assertEquals("taosdata", conf.getString("password"));
-            Assert.assertEquals("weather", conf.getString("table[0]"));
-            Assert.assertEquals("jdbc:TAOS-RS://master:6041/test", conf.getString("jdbcUrl"));
-        }
+        Assert.assertEquals(1, configurationList.size());
+        Configuration conf = configurationList.get(0);
+        Assert.assertEquals("root", conf.getString("username"));
+        Assert.assertEquals("taosdata", conf.getString("password"));
+        Assert.assertEquals("weather", conf.getString("table[0]"));
+        Assert.assertEquals("jdbc:TAOS-RS://master:6041/test", conf.getString("jdbcUrl"));
+
+    }
+
+    @Test
+    public void jobSplit_case02() {
+        // given
+        TDengineReader.Job job = new TDengineReader.Job();
+        Configuration configuration = Configuration.from("{" +
+                "\"username\": \"root\"," +
+                "\"password\": \"taosdata\"," +
+                "\"connection\": [{\"table\":[\"weather\"],\"jdbcUrl\":\"jdbc:TAOS-RS://master:6041/test\"}]," +
+                "\"column\": [\"ts\",\"current\",\"voltage\",\"phase\"]," +
+                "}");
+        job.setPluginJobConf(configuration);
+
+        // when
+        job.init();
+        List<Configuration> configurationList = job.split(1);
+
+        // assert
+        Assert.assertEquals(1, configurationList.size());
+        Configuration conf = configurationList.get(0);
+        Assert.assertEquals("root", conf.getString("username"));
+        Assert.assertEquals("taosdata", conf.getString("password"));
+        Assert.assertEquals("weather", conf.getString("table[0]"));
+        Assert.assertEquals("jdbc:TAOS-RS://master:6041/test", conf.getString("jdbcUrl"));
     }
 
 }

tdenginereader/src/test/resources/t2dm.json

@@ -0,0 +1,50 @@
+{
+  "job": {
+    "content": [
+      {
+        "reader": {
+          "name": "tdenginereader",
+          "parameter": {
+            "username": "root",
+            "password": "taosdata",
+            "column": [
+              "*"
+            ],
+            "connection": [
+              {
+                "table": [
+                  "stb1"
+                ],
+                "jdbcUrl": "jdbc:TAOS-RS://192.168.56.105:6041/db1"
+              }
+            ]
+          }
+        },
+        "writer": {
+          "name": "rdbmswriter",
+          "parameter": {
+            "connection": [
+              {
+                "table": [
+                  "stb2"
+                ],
+                "jdbcUrl": "jdbc:dm://192.168.0.72:5236"
+              }
+            ],
+            "username": "TESTUSER",
+            "password": "test123456",
+            "table": "stb2",
+            "column": [
+              "*"
+            ]
+          }
+        }
+      }
+    ],
+    "setting": {
+      "speed": {
+        "channel": 1
+      }
+    }
+  }
+}

tdenginewriter/pom.xml

@@ -75,14 +75,14 @@
             <scope>test</scope>
         </dependency>
         <!-- 添加 dm8 jdbc jar 包依赖-->
-        <dependency>
-            <groupId>com.dameng</groupId>
-            <artifactId>dm-jdbc</artifactId>
-            <version>1.8</version>
-            <scope>system</scope>
-            <systemPath>${project.basedir}/src/test/resources/DmJdbcDriver18.jar
-            </systemPath>
-        </dependency>
+<!--        <dependency>-->
+<!--            <groupId>com.dameng</groupId>-->
+<!--            <artifactId>dm-jdbc</artifactId>-->
+<!--            <version>1.8</version>-->
+<!--            <scope>system</scope>-->
+<!--            <systemPath>${project.basedir}/src/test/resources/DmJdbcDriver18.jar-->
+<!--            </systemPath>-->
+<!--        </dependency>-->
 
     </dependencies>
 

tdenginewriter/src/main/java/com/alibaba/datax/plugin/writer/tdenginewriter/DefaultDataHandler.java

@@ -19,6 +19,15 @@ import java.util.Date;
 import java.util.stream.Collectors;
 
 public class DefaultDataHandler implements DataHandler {
+    static {
+        try {
+            Class.forName("com.taosdata.jdbc.TSDBDriver");
+            Class.forName("com.taosdata.jdbc.rs.RestfulDriver");
+        } catch (ClassNotFoundException e) {
+            e.printStackTrace();
+        }
+    }
+
     private static final Logger LOG = LoggerFactory.getLogger(DefaultDataHandler.class);
 
     private String username;
@@ -62,6 +71,7 @@ public class DefaultDataHandler implements DataHandler {
         int count = 0;
         int affectedRows = 0;
 
+
         try (Connection conn = DriverManager.getConnection(jdbcUrl, username, password)) {
             LOG.info("connection[ jdbcUrl: " + jdbcUrl + ", username: " + username + "] established.");
             // prepare table_name -> table_meta

tdenginewriter/src/main/java/com/alibaba/datax/plugin/writer/tdenginewriter/Key.java

@@ -9,4 +9,7 @@ public class Key {
     public static final String JDBC_URL = "jdbcUrl";
     public static final String COLUMN = "column";
     public static final String IGNORE_TAGS_UNMATCHED = "ignoreTagsUnmatched";
+
+    public static final String BEGIN_DATETIME = "beginDateTime";
+    public static final String END_DATETIME = "endDateTime";
 }

tdenginewriter/src/main/java/com/alibaba/datax/plugin/writer/tdenginewriter/SchemaManager.java

@@ -17,7 +17,7 @@ public class SchemaManager {
     private final Connection conn;
     private TimestampPrecision precision;
 
-    SchemaManager(Connection conn) {
+    public SchemaManager(Connection conn) {
         this.conn = conn;
     }
 

tdenginewriter/src/main/java/com/alibaba/datax/plugin/writer/tdenginewriter/TDengineWriter.java

@@ -25,7 +25,7 @@ public class TDengineWriter extends Writer {
             this.originalConfig = super.getPluginJobConf();
             this.originalConfig.set(PEER_PLUGIN_NAME, getPeerPluginName());
 
-            // check user
+            // check username
             String user = this.originalConfig.getString(Key.USERNAME);
             if (StringUtils.isBlank(user))
                 throw DataXException.asDataXException(TDengineWriterErrorCode.REQUIRED_VALUE, "The parameter ["

tdenginewriter/src/test/java/com/alibaba/datax/plugin/writer/tdenginewriter/DM2TDengineTest.java

@@ -99,7 +99,7 @@ public class DM2TDengineTest {
     }
 
     @Before
-    public void before() throws SQLException, ClassNotFoundException {
+    public void before() throws SQLException {
         SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
         long ts = System.currentTimeMillis();
 

tdenginewriter/src/test/resources/incremental_sync/clean_env.sh

@@ -0,0 +1,9 @@
+#!/bin/bash
+
+datax_home_dir=$(dirname $(readlink -f "$0"))
+
+curl -H 'Authorization: Basic cm9vdDp0YW9zZGF0YQ==' -d 'drop table if exists db2.stb2;' 192.168.1.93:6041/rest/sql
+curl -H 'Authorization: Basic cm9vdDp0YW9zZGF0YQ==' -d 'create table if not exists db2.stb2 (`ts` TIMESTAMP,`f2` SMALLINT,`f4` BIGINT,`f5` FLOAT,`f6` DOUBLE,`f7` DOUBLE,`f8` BOOL,`f9` NCHAR(100),`f10` NCHAR(200)) TAGS (`f1` TINYINT,`f3` INT);' 192.168.1.93:6041/rest/sql
+
+rm -f ${datax_home_dir}/log/*
+rm -f ${datax_home_dir}/job/*.csv

tdenginewriter/src/test/resources/incremental_sync/dm2t-update.json

@@ -0,0 +1,63 @@
+{
+  "job": {
+    "content": [
+      {
+        "reader": {
+          "name": "rdbmsreader",
+          "parameter": {
+            "username": "TESTUSER",
+            "password": "test123456",
+            "connection": [
+              {
+                "querySql": [
+                  "select concat(concat(concat('t', f1), '_'),f3) as tbname,* from stb1"
+                ],
+                "jdbcUrl": [
+                  "jdbc:dm://192.168.0.72:5236"
+                ]
+              }
+            ],
+            "where": "1=1",
+            "fetchSize": 1024
+          }
+        },
+        "writer": {
+          "name": "tdenginewriter",
+          "parameter": {
+            "username": "root",
+            "password": "taosdata",
+            "column": [
+              "tbname",
+              "ts",
+              "f1",
+              "f2",
+              "f3",
+              "f4",
+              "f5",
+              "f6",
+              "f7",
+              "f8",
+              "f9",
+              "f10"
+            ],
+            "connection": [
+              {
+                "table": [
+                  "stb2"
+                ],
+                "jdbcUrl": "jdbc:TAOS-RS://192.168.1.93:6041/db2"
+              }
+            ],
+            "batchSize": 1000,
+            "ignoreTagsUnmatched": true
+          }
+        }
+      }
+    ],
+    "setting": {
+      "speed": {
+        "channel": 1
+      }
+    }
+  }
+}

tdenginewriter/src/test/resources/incremental_sync/dm2t_sync.sh

@@ -0,0 +1,57 @@
+#!/bin/bash
+
+set -e
+#set -x
+
+datax_home_dir=$(dirname $(readlink -f "$0"))
+table_name="stb1"
+update_key="ts"
+
+while getopts "hd:t:" arg; do
+  case $arg in
+  d)
+    datax_home_dir=$(echo $OPTARG)
+    ;;
+  v)
+    table_name=$(echo $OPTARG)
+    ;;
+  h)
+    echo "Usage: $(basename $0) -d [datax_home_dir] -t [table_name] -k [update_key]"
+    echo "                    -h help"
+    exit 0
+    ;;
+  ?) #unknow option
+    echo "unkonw argument"
+    exit 1
+    ;;
+  esac
+done
+
+if [[ -e ${datax_home_dir}/job/${table_name}.csv ]]; then
+  MAX_TIME=$(cat ${datax_home_dir}/job/${table_name}.csv)
+else
+  MAX_TIME="null"
+fi
+current_datetime=$(date +"%Y-%m-%d %H:%M:%S")
+current_timestamp=$(date +%s)
+
+if [ "$MAX_TIME" != "null" ]; then
+  WHERE="${update_key} >= '$MAX_TIME' and ${update_key} < '$current_datetime'"
+  sed "s/1=1/$WHERE/g" ${datax_home_dir}/job/dm2t-update.json >${datax_home_dir}/job/dm2t_${current_timestamp}.json
+  echo "incremental data synchronization, from '${MAX_TIME}' to '${current_datetime}'"
+  python ${datax_home_dir}/bin/datax.py ${datax_home_dir}/job/dm2t_${current_timestamp}.json 1> /dev/null 2>&1
+else
+  echo "full data synchronization, to '${current_datetime}'"
+  python ${datax_home_dir}/bin/datax.py ${datax_home_dir}/job/dm2t-update.json 1> /dev/null 2>&1
+fi
+
+if [[ $? -ne 0 ]]; then
+  echo "datax migration job falied"
+else
+  echo ${current_datetime} >$datax_home_dir/job/${table_name}.csv
+  echo "datax migration job success"
+fi
+
+rm -rf ${datax_home_dir}/job/dm2t_${current_timestamp}.json
+
+#while true; do ./dm2t_sync.sh; sleep 5s; done

tdenginewriter/src/test/resources/incremental_sync/upload.sh

@@ -0,0 +1,5 @@
+#!/bin/bash
+
+scp dm2t-update.json root@192.168.56.105:/root/workspace/tmp/datax/job
+scp dm2t_sync.sh root@192.168.56.105:/root/workspace/tmp/datax
+scp clean_env.sh root@192.168.56.105:/root/workspace/tmp/datax