diff --git a/.travlis.yml b/.travlis.yml
new file mode 100644
index 00000000..b08b39cb
--- /dev/null
+++ b/.travlis.yml
@@ -0,0 +1,23 @@
+language: java
+sudo: false # faster builds
+
+jdk:
+ - openjdk11
+ - oraclejdk11
+ - openjdk8
+ - oraclejdk8
+
+cache:
+ directories:
+ - $HOME/.m2
+
+install: true
+
+script:
+ - travis_wait 30 ./mvnw clean install -DskipTests=true -Dcheckstyle.skip=true -Drat.skip=true -Dmaven.javadoc.skip=true
+
+after_success:
+ - bash <(curl -s https://codecov.io/bash)
+
+after_failure:
+ - if [ -f canal.log ]; then echo "------TAIL of canal.log------"; tail -n 1000 canal.log; echo "------END of canal.log------"; fi
diff --git a/README.md b/README.md
index 8941c8fb..36dead59 100644
--- a/README.md
+++ b/README.md
@@ -82,7 +82,7 @@ See the wiki page for : wiki文
1. canal整体交互协议设计上使用了protobuf3.0,理论上可以支持绝大部分的多语言场景,欢迎大家提交多客户端的PR
* canal java客户端: https://github.com/alibaba/canal/wiki/ClientExample
- * canal c#客户端开源项目地址: https://github.com/CanalSharp/CanalSharp
+ * canal c#客户端开源项目地址: https://github.com/dotnetcore/CanalSharp
* canal go客户端开源项目地址: https://github.com/CanalClient/canal-go
2. canal作为MySQL binlog的增量获取工具,可以将数据投递到MQ系统中,比如Kafka/RocketMQ,可以借助于MQ的多语言能力
* 参考文档: [Canal Kafka/RocketMQ QuickStart](https://github.com/alibaba/canal/wiki/Canal-Kafka-RocketMQ-QuickStart)
@@ -117,7 +117,7 @@ See the wiki page for : wiki文
最新更新
- canal发布重大版本更新1.1.0,具体releaseNode参考:https://github.com/alibaba/canal/releases/tag/canal-1.1.0
-- canal c#客户端开源项目地址: https://github.com/CanalClient/CanalSharp ,推荐!
+- canal c#客户端开源项目地址: https://github.com/dotnetcore/CanalSharp ,推荐!
- canal QQ讨论群已经建立,群号:161559791 ,欢迎加入进行技术讨论。
- canal消费端项目开源: Otter(分布式数据库同步系统),地址:https://github.com/alibaba/otter
diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/config/YmlConfigBinder.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/config/YmlConfigBinder.java
index 705d7df3..837b1ab4 100644
--- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/config/YmlConfigBinder.java
+++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/config/YmlConfigBinder.java
@@ -84,7 +84,7 @@ public class YmlConfigBinder {
try {
byte[] contentBytes;
if (charset == null) {
- contentBytes = content.getBytes();
+ contentBytes = content.getBytes("UTF-8");
} else {
contentBytes = content.getBytes(charset);
}
diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/config/bind/RelaxedConversionService.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/config/bind/RelaxedConversionService.java
index 00a288b6..ab8e5506 100644
--- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/config/bind/RelaxedConversionService.java
+++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/config/bind/RelaxedConversionService.java
@@ -89,7 +89,7 @@ class RelaxedConversionService implements ConversionService {
return new RelaxedConversionService.StringToEnumIgnoringCaseConverterFactory.StringToEnum(enumType);
}
- private class StringToEnum implements Converter {
+ private static class StringToEnum implements Converter {
private final Class enumType;
diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/config/common/YamlProcessor.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/config/common/YamlProcessor.java
index 819c99fc..c8dbdbaf 100644
--- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/config/common/YamlProcessor.java
+++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/config/common/YamlProcessor.java
@@ -191,6 +191,7 @@ public abstract class YamlProcessor {
}
private boolean process(Map map, MatchCallback callback) {
+ @SuppressWarnings("serial")
Properties properties = new Properties() {
@Override
diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/ExtensionLoader.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/ExtensionLoader.java
index 41d1864e..bb73cb54 100644
--- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/ExtensionLoader.java
+++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/ExtensionLoader.java
@@ -6,6 +6,7 @@ import java.io.InputStreamReader;
import java.net.MalformedURLException;
import java.net.URL;
import java.net.URLClassLoader;
+import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.Arrays;
import java.util.Enumeration;
@@ -27,7 +28,8 @@ import org.slf4j.LoggerFactory;
*/
public class ExtensionLoader {
- private static final Logger logger = LoggerFactory.getLogger(ExtensionLoader.class);
+ private static final Logger logger = LoggerFactory
+ .getLogger(ExtensionLoader.class);
private static final String SERVICES_DIRECTORY = "META-INF/services/";
@@ -35,7 +37,8 @@ public class ExtensionLoader {
private static final String DEFAULT_CLASSLOADER_POLICY = "internal";
- private static final Pattern NAME_SEPARATOR = Pattern.compile("\\s*[,]+\\s*");
+ private static final Pattern NAME_SEPARATOR = Pattern
+ .compile("\\s*[,]+\\s*");
private static final ConcurrentMap, ExtensionLoader>> EXTENSION_LOADERS = new ConcurrentHashMap<>();
@@ -171,7 +174,8 @@ public class ExtensionLoader {
return instance;
} catch (Throwable t) {
throw new IllegalStateException("Extension instance(name: " + name + ", class: " + type
- + ") could not be instantiated: " + t.getMessage(), t);
+ + ") could not be instantiated: " + t.getMessage(),
+ t);
}
}
@@ -191,7 +195,8 @@ public class ExtensionLoader {
return instance;
} catch (Throwable t) {
throw new IllegalStateException("Extension instance(name: " + name + ", class: " + type
- + ") could not be instantiated: " + t.getMessage(), t);
+ + ") could not be instantiated: " + t.getMessage(),
+ t);
}
}
@@ -233,7 +238,11 @@ public class ExtensionLoader {
// button.
jarPath = jarPath.replaceAll("/classes/.*", "/classes/");
}
- return Paths.get(jarPath).getParent().toString(); // Paths - from java 8
+ Path path = Paths.get(jarPath).getParent(); // Paths - from java 8
+ if (path != null) {
+ return path.toString();
+ }
+ return null;
}
private Map> loadExtensionClasses() {
@@ -330,12 +339,10 @@ public class ExtensionLoader {
// Class.forName(line, true,
// classLoader);
if (!type.isAssignableFrom(clazz)) {
- throw new IllegalStateException("Error when load extension class(interface: "
- + type
- + ", class line: "
- + clazz.getName()
- + "), class "
- + clazz.getName()
+ throw new IllegalStateException(
+ "Error when load extension class(interface: " + type
+ + ", class line: " + clazz.getName()
+ + "), class " + clazz.getName()
+ "is not subtype of interface.");
} else {
try {
@@ -353,9 +360,9 @@ public class ExtensionLoader {
extensionClasses.put(n, clazz);
} else if (c != clazz) {
cachedNames.remove(clazz);
- throw new IllegalStateException("Duplicate extension "
- + type.getName()
- + " name " + n + " on "
+ throw new IllegalStateException(
+ "Duplicate extension " + type.getName() + " name "
+ + n + " on "
+ c.getName() + " and "
+ clazz.getName());
}
@@ -365,12 +372,9 @@ public class ExtensionLoader {
}
}
} catch (Throwable t) {
- IllegalStateException e = new IllegalStateException("Failed to load extension class(interface: "
- + type
- + ", class line: "
- + line
- + ") in "
- + url
+ IllegalStateException e = new IllegalStateException(
+ "Failed to load extension class(interface: " + type + ", class line: "
+ + line + ") in " + url
+ ", cause: "
+ t.getMessage(),
t);
@@ -385,13 +389,15 @@ public class ExtensionLoader {
}
} catch (Throwable t) {
logger.error("Exception when load extension class(interface: " + type + ", class file: " + url
- + ") in " + url, t);
+ + ") in " + url,
+ t);
}
} // end of while urls
}
} catch (Throwable t) {
- logger.error("Exception when load extension class(interface: " + type + ", description file: " + fileName
- + ").", t);
+ logger.error(
+ "Exception when load extension class(interface: " + type + ", description file: " + fileName + ").",
+ t);
}
}
diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/JdbcTypeUtil.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/JdbcTypeUtil.java
index ebcb4ec0..4dd061c6 100644
--- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/JdbcTypeUtil.java
+++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/JdbcTypeUtil.java
@@ -155,9 +155,11 @@ public class JdbcTypeUtil {
} else {
res = null;
}
+ break;
case Types.CLOB:
default:
res = value;
+ break;
}
return res;
} catch (Exception e) {
diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Result.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Result.java
index 6e9a9954..a0e8f905 100644
--- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Result.java
+++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Result.java
@@ -12,10 +12,10 @@ import java.util.Date;
public class Result implements Serializable {
private static final long serialVersionUID = -3276409502352405716L;
- public Integer code = 20000;
- public Object data;
- public String message;
- public Date sysTime;
+ private Integer code = 20000;
+ private Object data;
+ private String message;
+ private Date sysTime;
public static Result createSuccess(String message) {
Result result = new Result();
diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Util.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Util.java
index f4878a47..78f3de39 100644
--- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Util.java
+++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Util.java
@@ -2,11 +2,17 @@ package com.alibaba.otter.canal.client.adapter.support;
import java.io.File;
import java.net.URL;
-import java.sql.*;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
import java.time.*;
import java.time.format.DateTimeFormatter;
import java.util.Date;
import java.util.TimeZone;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.function.Function;
@@ -29,38 +35,16 @@ public class Util {
/**
* 通过DS执行sql
*/
- public static Object sqlRS(DataSource ds, String sql, Function fun) throws SQLException {
- Connection conn = null;
- Statement stmt = null;
- ResultSet rs = null;
- try {
- conn = ds.getConnection();
- stmt = conn.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
+ public static Object sqlRS(DataSource ds, String sql, Function fun) {
+ try (Connection conn = ds.getConnection();
+ Statement stmt = conn.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);) {
stmt.setFetchSize(Integer.MIN_VALUE);
- rs = stmt.executeQuery(sql);
- return fun.apply(rs);
- } finally {
- if (rs != null) {
- try {
- rs.close();
- } catch (SQLException e) {
- logger.error(e.getMessage(), e);
- }
- }
- if (stmt != null) {
- try {
- stmt.close();
- } catch (SQLException e) {
- logger.error(e.getMessage(), e);
- }
- }
- if (conn != null) {
- try {
- conn.close();
- } catch (SQLException e) {
- logger.error(e.getMessage(), e);
- }
+ try (ResultSet rs = stmt.executeQuery(sql);) {
+ return fun.apply(rs);
}
+ } catch (Exception e) {
+ logger.error("sqlRs has error, sql: {} ", sql);
+ throw new RuntimeException(e);
}
}
@@ -72,29 +56,10 @@ public class Util {
* @param consumer 回调方法
*/
public static void sqlRS(Connection conn, String sql, Consumer consumer) {
- Statement stmt = null;
- ResultSet rs = null;
- try {
- stmt = conn.createStatement();
- rs = stmt.executeQuery(sql);
+ try (Statement stmt = conn.createStatement(); ResultSet rs = stmt.executeQuery(sql)) {
consumer.accept(rs);
} catch (SQLException e) {
logger.error(e.getMessage(), e);
- } finally {
- if (rs != null) {
- try {
- rs.close();
- } catch (SQLException e) {
- logger.error(e.getMessage(), e);
- }
- }
- if (stmt != null) {
- try {
- stmt.close();
- } catch (SQLException e) {
- logger.error(e.getMessage(), e);
- }
- }
}
}
@@ -144,7 +109,41 @@ public class Util {
return column;
}
- private static String timeZone; // 当前时区
+ public static ThreadPoolExecutor newFixedThreadPool(int nThreads, long keepAliveTime) {
+ return new ThreadPoolExecutor(nThreads,
+ nThreads,
+ keepAliveTime,
+ TimeUnit.MILLISECONDS,
+ new SynchronousQueue<>(),
+ (r, exe) -> {
+ if (!exe.isShutdown()) {
+ try {
+ exe.getQueue().put(r);
+ } catch (InterruptedException e) {
+ // ignore
+ }
+ }
+ });
+ }
+
+ public static ThreadPoolExecutor newSingleThreadExecutor(long keepAliveTime) {
+ return new ThreadPoolExecutor(1,
+ 1,
+ keepAliveTime,
+ TimeUnit.MILLISECONDS,
+ new SynchronousQueue<>(),
+ (r, exe) -> {
+ if (!exe.isShutdown()) {
+ try {
+ exe.getQueue().put(r);
+ } catch (InterruptedException e) {
+ // ignore
+ }
+ }
+ });
+ }
+
+ public final static String timeZone; // 当前时区
private static DateTimeZone dateTimeZone;
static {
@@ -265,7 +264,7 @@ public class Util {
LocalDateTime localDateTime = LocalDateTime.of(localDate, localTime);
return Date.from(localDateTime.atZone(ZoneId.systemDefault()).toInstant());
}
- } catch (Exception e) {
+ } catch (Throwable e) {
logger.error(e.getMessage(), e);
}
diff --git a/client-adapter/elasticsearch/pom.xml b/client-adapter/elasticsearch/pom.xml
index 90c64f80..d06eed6f 100644
--- a/client-adapter/elasticsearch/pom.xml
+++ b/client-adapter/elasticsearch/pom.xml
@@ -40,6 +40,12 @@
4.12
test
+
+ mysql
+ mysql-connector-java
+ 5.1.40
+ test
+
diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java
index c2e899e7..cae10145 100644
--- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java
+++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java
@@ -133,8 +133,8 @@ public class ESAdapter implements OuterAdapter {
esSyncService = new ESSyncService(esTemplate);
esConfigMonitor = new ESConfigMonitor();
- esConfigMonitor.init(this);
- } catch (Exception e) {
+ esConfigMonitor.init(this, envProperties);
+ } catch (Throwable e) {
throw new RuntimeException(e);
}
}
diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfig.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfig.java
index 94341fd0..e77ecd47 100644
--- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfig.java
+++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfig.java
@@ -31,7 +31,7 @@ public class ESSyncConfig {
throw new NullPointerException("esMapping._type");
}
if (esMapping._id == null && esMapping.getPk() == null) {
- throw new NullPointerException("esMapping._id and esMapping.pk");
+ throw new NullPointerException("esMapping._id or esMapping.pk");
}
if (esMapping.sql == null) {
throw new NullPointerException("esMapping.sql");
@@ -80,23 +80,23 @@ public class ESSyncConfig {
public static class ESMapping {
- private String _index;
- private String _type;
- private String _id;
- private boolean upsert = false;
- private String pk;
- // private String parent;
- private String sql;
+ private String _index;
+ private String _type;
+ private String _id;
+ private boolean upsert = false;
+ private String pk;
+ private Map relations = new LinkedHashMap<>();
+ private String sql;
// 对象字段, 例: objFields:
// - _labels: array:;
- private Map objFields = new LinkedHashMap<>();
- private List skips = new ArrayList<>();
- private int commitBatch = 1000;
- private String etlCondition;
- private boolean syncByTimestamp = false; // 是否按时间戳定时同步
- private Long syncInterval; // 同步时间间隔
+ private Map objFields = new LinkedHashMap<>();
+ private List skips = new ArrayList<>();
+ private int commitBatch = 1000;
+ private String etlCondition;
+ private boolean syncByTimestamp = false; // 是否按时间戳定时同步
+ private Long syncInterval; // 同步时间间隔
- private SchemaItem schemaItem; // sql解析结果模型
+ private SchemaItem schemaItem; // sql解析结果模型
public String get_index() {
return _index;
@@ -154,6 +154,14 @@ public class ESSyncConfig {
this.skips = skips;
}
+ public Map getRelations() {
+ return relations;
+ }
+
+ public void setRelations(Map relations) {
+ this.relations = relations;
+ }
+
public String getSql() {
return sql;
}
@@ -202,4 +210,26 @@ public class ESSyncConfig {
this.schemaItem = schemaItem;
}
}
+
+ public static class RelationMapping {
+
+ private String name;
+ private String parent;
+
+ public String getName() {
+ return name;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ public String getParent() {
+ return parent;
+ }
+
+ public void setParent(String parent) {
+ this.parent = parent;
+ }
+ }
}
diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SchemaItem.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SchemaItem.java
index 4d45c525..bad97bd4 100644
--- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SchemaItem.java
+++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SchemaItem.java
@@ -278,12 +278,13 @@ public class SchemaItem {
if (relationSelectFieldItems == null) {
synchronized (SchemaItem.class) {
if (relationSelectFieldItems == null) {
- relationSelectFieldItems = new ArrayList<>();
+ List relationSelectFieldItemsTmp = new ArrayList<>();
for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
if (fieldItem.getOwners().contains(getAlias())) {
- relationSelectFieldItems.add(fieldItem);
+ relationSelectFieldItemsTmp.add(fieldItem);
}
}
+ relationSelectFieldItems = relationSelectFieldItemsTmp;
}
}
}
diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/monitor/ESConfigMonitor.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/monitor/ESConfigMonitor.java
index 566a9fe0..23f7bfb5 100644
--- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/monitor/ESConfigMonitor.java
+++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/monitor/ESConfigMonitor.java
@@ -3,6 +3,7 @@ package com.alibaba.otter.canal.client.adapter.es.monitor;
import java.io.File;
import java.util.HashMap;
import java.util.Map;
+import java.util.Properties;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@@ -12,9 +13,9 @@ import org.apache.commons.io.monitor.FileAlterationMonitor;
import org.apache.commons.io.monitor.FileAlterationObserver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.yaml.snakeyaml.Yaml;
import com.alibaba.druid.pool.DruidDataSource;
+import com.alibaba.otter.canal.client.adapter.config.YmlConfigBinder;
import com.alibaba.otter.canal.client.adapter.es.ESAdapter;
import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig;
import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem;
@@ -31,10 +32,13 @@ public class ESConfigMonitor {
private ESAdapter esAdapter;
+ private Properties envProperties;
+
private FileAlterationMonitor fileMonitor;
- public void init(ESAdapter esAdapter) {
+ public void init(ESAdapter esAdapter, Properties envProperties) {
this.esAdapter = esAdapter;
+ this.envProperties = envProperties;
File confDir = Util.getConfDirPath(adapterName);
try {
FileAlterationObserver observer = new FileAlterationObserver(confDir,
@@ -65,11 +69,13 @@ public class ESConfigMonitor {
try {
// 加载新增的配置文件
String configContent = MappingConfigsLoader.loadConfig(adapterName + File.separator + file.getName());
- ESSyncConfig config = new Yaml().loadAs(configContent, ESSyncConfig.class);
- config.validate();
- addConfigToCache(file, config);
-
- logger.info("Add a new es mapping config: {} to canal adapter", file.getName());
+ ESSyncConfig config = YmlConfigBinder
+ .bindYmlToObj(null, configContent, ESSyncConfig.class, null, envProperties);
+ if (config != null) {
+ config.validate();
+ addConfigToCache(file, config);
+ logger.info("Add a new es mapping config: {} to canal adapter", file.getName());
+ }
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
@@ -88,7 +94,11 @@ public class ESConfigMonitor {
onFileDelete(file);
return;
}
- ESSyncConfig config = new Yaml().loadAs(configContent, ESSyncConfig.class);
+ ESSyncConfig config = YmlConfigBinder
+ .bindYmlToObj(null, configContent, ESSyncConfig.class, null, envProperties);
+ if (config == null) {
+ return;
+ }
config.validate();
if (esAdapter.getEsSyncConfig().containsKey(file.getName())) {
deleteConfigFromCache(file);
diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESEtlService.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESEtlService.java
index 827d9460..45b0e60e 100644
--- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESEtlService.java
+++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESEtlService.java
@@ -1,25 +1,23 @@
package com.alibaba.otter.canal.client.adapter.es.service;
-import java.util.ArrayList;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
+import java.sql.SQLException;
+import java.util.*;
import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import javax.sql.DataSource;
+import org.apache.commons.lang.StringUtils;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
+import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
-import org.elasticsearch.action.search.SearchType;
+import org.elasticsearch.action.update.UpdateRequestBuilder;
import org.elasticsearch.client.transport.TransportClient;
-import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHit;
@@ -30,10 +28,10 @@ import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig;
import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig.ESMapping;
import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem.FieldItem;
-import com.alibaba.otter.canal.client.adapter.es.support.ESSyncUtil;
import com.alibaba.otter.canal.client.adapter.es.support.ESTemplate;
import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
import com.alibaba.otter.canal.client.adapter.support.EtlResult;
+import com.alibaba.otter.canal.client.adapter.support.Util;
import com.google.common.base.Joiner;
/**
@@ -100,7 +98,7 @@ public class ESEtlService {
// 获取总数
String countSql = "SELECT COUNT(1) FROM ( " + sql + ") _CNT ";
- long cnt = (Long) ESSyncUtil.sqlRS(dataSource, countSql, rs -> {
+ long cnt = (Long) Util.sqlRS(dataSource, countSql, rs -> {
Long count = null;
try {
if (rs.next()) {
@@ -116,8 +114,7 @@ public class ESEtlService {
if (cnt >= 10000) {
int threadCount = 3; // 从配置读取默认为3
long perThreadCnt = cnt / threadCount;
- ExecutorService executor = Executors.newFixedThreadPool(threadCount);
- List> futures = new ArrayList<>(threadCount);
+ ExecutorService executor = Util.newFixedThreadPool(threadCount, 5000L);
for (int i = 0; i < threadCount; i++) {
long offset = i * perThreadCnt;
Long size = null;
@@ -130,16 +127,13 @@ public class ESEtlService {
} else {
sqlFinal = sql + " LIMIT " + offset + "," + cnt;
}
- Future future = executor
- .submit(() -> executeSqlImport(dataSource, sqlFinal, mapping, impCount, errMsg));
- futures.add(future);
- }
-
- for (Future future : futures) {
- future.get();
+ executor.execute(() -> executeSqlImport(dataSource, sqlFinal, mapping, impCount, errMsg));
}
executor.shutdown();
+ while (!executor.awaitTermination(3, TimeUnit.SECONDS)) {
+ // ignore
+ }
} else {
executeSqlImport(dataSource, sql, mapping, impCount, errMsg);
}
@@ -176,7 +170,7 @@ public class ESEtlService {
private boolean executeSqlImport(DataSource ds, String sql, ESMapping mapping, AtomicLong impCount,
List errMsg) {
try {
- ESSyncUtil.sqlRS(ds, sql, rs -> {
+ Util.sqlRS(ds, sql, rs -> {
int count = 0;
try {
BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk();
@@ -184,39 +178,73 @@ public class ESEtlService {
long batchBegin = System.currentTimeMillis();
while (rs.next()) {
Map esFieldData = new LinkedHashMap<>();
+ Object idVal = null;
for (FieldItem fieldItem : mapping.getSchemaItem().getSelectFields().values()) {
- // 如果是主键字段则不插入
- if (fieldItem.getFieldName().equals(mapping.get_id())) {
- continue;
- }
-
String fieldName = fieldItem.getFieldName();
if (mapping.getSkips().contains(fieldName)) {
continue;
}
- Object val = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName);
- esFieldData.put(fieldName, val);
+ // 如果是主键字段则不插入
+ if (fieldItem.getFieldName().equals(mapping.get_id())) {
+ idVal = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName);
+ } else {
+ Object val = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName);
+ esFieldData.put(Util.cleanColumn(fieldName), val);
+ }
+
}
- Object idVal = null;
- if (mapping.get_id() != null) {
- idVal = rs.getObject(mapping.get_id());
+
+ if (!mapping.getRelations().isEmpty()) {
+ mapping.getRelations().forEach((relationField, relationMapping) -> {
+ Map relations = new HashMap<>();
+ relations.put("name", relationMapping.getName());
+ if (StringUtils.isNotEmpty(relationMapping.getParent())) {
+ FieldItem parentFieldItem = mapping.getSchemaItem()
+ .getSelectFields()
+ .get(relationMapping.getParent());
+ Object parentVal;
+ try {
+ parentVal = esTemplate.getValFromRS(mapping,
+ rs,
+ parentFieldItem.getFieldName(),
+ parentFieldItem.getFieldName());
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ if (parentVal != null) {
+ relations.put("parent", parentVal.toString());
+ esFieldData.put("$parent_routing", parentVal.toString());
+
+ }
+ }
+ esFieldData.put(Util.cleanColumn(relationField), relations);
+ });
}
if (idVal != null) {
+ String parentVal = (String) esFieldData.remove("$parent_routing");
if (mapping.isUpsert()) {
- bulkRequestBuilder.add(transportClient
+ UpdateRequestBuilder updateRequestBuilder = transportClient
.prepareUpdate(mapping.get_index(), mapping.get_type(), idVal.toString())
.setDoc(esFieldData)
- .setDocAsUpsert(true));
+ .setDocAsUpsert(true);
+ if (StringUtils.isNotEmpty(parentVal)) {
+ updateRequestBuilder.setRouting(parentVal);
+ }
+ bulkRequestBuilder.add(updateRequestBuilder);
} else {
- bulkRequestBuilder.add(transportClient
+ IndexRequestBuilder indexRequestBuilder = transportClient
.prepareIndex(mapping.get_index(), mapping.get_type(), idVal.toString())
- .setSource(esFieldData));
+ .setSource(esFieldData);
+ if (StringUtils.isNotEmpty(parentVal)) {
+ indexRequestBuilder.setRouting(parentVal);
+ }
+ bulkRequestBuilder.add(indexRequestBuilder);
}
} else {
- idVal = rs.getObject(mapping.getPk());
+ idVal = esFieldData.get(mapping.getPk());
SearchResponse response = transportClient.prepareSearch(mapping.get_index())
.setTypes(mapping.get_type())
.setQuery(QueryBuilders.termQuery(mapping.getPk(), idVal))
diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java
index 0ba4d442..ace00cba 100644
--- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java
+++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java
@@ -7,6 +7,7 @@ import java.util.Map;
import javax.sql.DataSource;
+import com.alibaba.otter.canal.client.adapter.support.Util;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -43,7 +44,7 @@ public class ESSyncService {
long begin = System.currentTimeMillis();
if (esSyncConfigs != null) {
if (logger.isTraceEnabled()) {
- logger.trace("Destination: {}, database:{}, table:{}, type:{}, effect index count: {}",
+ logger.trace("Destination: {}, database:{}, table:{}, type:{}, affected index count: {}",
dml.getDestination(),
dml.getDatabase(),
dml.getTable(),
@@ -65,7 +66,7 @@ public class ESSyncService {
}
}
if (logger.isTraceEnabled()) {
- logger.trace("Sync elapsed time: {} ms, effect index count:{}, destination: {}",
+ logger.trace("Sync elapsed time: {} ms, affected indexes count:{}, destination: {}",
(System.currentTimeMillis() - begin),
esSyncConfigs.size(),
dml.getDestination());
@@ -74,7 +75,7 @@ public class ESSyncService {
StringBuilder configIndexes = new StringBuilder();
esSyncConfigs
.forEach(esSyncConfig -> configIndexes.append(esSyncConfig.getEsMapping().get_index()).append(" "));
- logger.debug("DML: {} \nEffect indexes: {}",
+ logger.debug("DML: {} \nAffected indexes: {}",
JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue),
configIndexes.toString());
}
@@ -166,7 +167,7 @@ public class ESSyncService {
data,
fieldItem.getFieldName(),
fieldItem.getColumn().getColumnName());
- esFieldData.put(fieldItem.getFieldName(), value);
+ esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value);
}
joinTableSimpleFieldOperation(config, dml, data, tableItem, esFieldData);
@@ -295,7 +296,7 @@ public class ESSyncService {
data,
fieldItem.getFieldName(),
fieldItem.getColumn().getColumnName());
- esFieldData.put(fieldItem.getFieldName(), value);
+ esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value);
}
}
joinTableSimpleFieldOperation(config, dml, data, tableItem, esFieldData);
@@ -407,7 +408,7 @@ public class ESSyncService {
// ------关联表简单字段更新为null------
Map esFieldData = new LinkedHashMap<>();
for (FieldItem fieldItem : tableItem.getRelationSelectFieldItems()) {
- esFieldData.put(fieldItem.getFieldName(), null);
+ esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), null);
}
joinTableSimpleFieldOperation(config, dml, data, tableItem, esFieldData);
} else {
@@ -464,7 +465,7 @@ public class ESSyncService {
mapping.get_index(),
sql.replace("\n", " "));
}
- ESSyncUtil.sqlRS(ds, sql, rs -> {
+ Util.sqlRS(ds, sql, rs -> {
try {
while (rs.next()) {
Map esFieldData = new LinkedHashMap<>();
@@ -500,7 +501,7 @@ public class ESSyncService {
mapping.get_index(),
sql.replace("\n", " "));
}
- ESSyncUtil.sqlRS(ds, sql, rs -> {
+ Util.sqlRS(ds, sql, rs -> {
try {
Map esFieldData = null;
if (mapping.getPk() != null) {
@@ -508,7 +509,7 @@ public class ESSyncService {
esTemplate.getESDataFromDmlData(mapping, data, esFieldData);
esFieldData.remove(mapping.getPk());
for (String key : esFieldData.keySet()) {
- esFieldData.put(key, null);
+ esFieldData.put(Util.cleanColumn(key), null);
}
}
while (rs.next()) {
@@ -601,7 +602,7 @@ public class ESSyncService {
mapping.get_index(),
sql.toString().replace("\n", " "));
}
- ESSyncUtil.sqlRS(ds, sql.toString(), rs -> {
+ Util.sqlRS(ds, sql.toString(), rs -> {
try {
while (rs.next()) {
Map esFieldData = new LinkedHashMap<>();
@@ -617,7 +618,7 @@ public class ESSyncService {
rs,
fieldItem.getFieldName(),
fieldItem.getColumn().getColumnName());
- esFieldData.put(fieldItem.getFieldName(), val);
+ esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), val);
break out;
}
}
@@ -628,7 +629,7 @@ public class ESSyncService {
rs,
fieldItem.getFieldName(),
fieldItem.getColumn().getColumnName());
- esFieldData.put(fieldItem.getFieldName(), val);
+ esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), val);
}
}
@@ -693,7 +694,7 @@ public class ESSyncService {
mapping.get_index(),
sql.toString().replace("\n", " "));
}
- ESSyncUtil.sqlRS(ds, sql.toString(), rs -> {
+ Util.sqlRS(ds, sql.toString(), rs -> {
try {
while (rs.next()) {
Map esFieldData = new LinkedHashMap<>();
@@ -724,7 +725,7 @@ public class ESSyncService {
rs,
fieldItem.getFieldName(),
fieldItem.getFieldName());
- esFieldData.put(fieldItem.getFieldName(), val);
+ esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), val);
break;
}
}
@@ -733,7 +734,7 @@ public class ESSyncService {
} else {
Object val = esTemplate
.getValFromRS(mapping, rs, fieldItem.getFieldName(), fieldItem.getFieldName());
- esFieldData.put(fieldItem.getFieldName(), val);
+ esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), val);
}
}
@@ -812,7 +813,7 @@ public class ESSyncService {
mapping.get_index(),
sql.replace("\n", " "));
}
- ESSyncUtil.sqlRS(ds, sql, rs -> {
+ Util.sqlRS(ds, sql, rs -> {
try {
while (rs.next()) {
Map esFieldData = new LinkedHashMap<>();
diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESSyncUtil.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESSyncUtil.java
index 84cb2f78..b49e3fcc 100644
--- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESSyncUtil.java
+++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESSyncUtil.java
@@ -3,12 +3,9 @@ package com.alibaba.otter.canal.client.adapter.es.support;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
-import java.sql.*;
+import java.sql.Blob;
+import java.sql.SQLException;
import java.util.*;
-import java.util.Date;
-import java.util.function.Function;
-
-import javax.sql.DataSource;
import org.apache.commons.codec.binary.Base64;
import org.joda.time.DateTime;
@@ -113,9 +110,9 @@ public class ESSyncUtil {
} else if (val instanceof java.sql.Timestamp) {
DateTime dateTime = new DateTime(((java.sql.Timestamp) val).getTime());
if (dateTime.getMillisOfSecond() != 0) {
- res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss.SSS+08:00");
+ res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss.SSS" + Util.timeZone);
} else {
- res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss+08:00");
+ res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss" + Util.timeZone);
}
} else if (val instanceof java.sql.Date || val instanceof Date) {
DateTime dateTime;
@@ -129,9 +126,9 @@ public class ESSyncUtil {
res = dateTime.toString("yyyy-MM-dd");
} else {
if (dateTime.getMillisOfSecond() != 0) {
- res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss.SSS+08:00");
+ res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss.SSS" + Util.timeZone);
} else {
- res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss+08:00");
+ res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss" + Util.timeZone);
}
}
} else if (val instanceof Long) {
@@ -140,9 +137,9 @@ public class ESSyncUtil {
&& dateTime.getMillisOfSecond() == 0) {
res = dateTime.toString("yyyy-MM-dd");
} else if (dateTime.getMillisOfSecond() != 0) {
- res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss.SSS+08:00");
+ res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss.SSS" + Util.timeZone);
} else {
- res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss+08:00");
+ res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss" + Util.timeZone);
}
} else if (val instanceof String) {
String v = ((String) val).trim();
@@ -153,9 +150,9 @@ public class ESSyncUtil {
if (date != null) {
DateTime dateTime = new DateTime(date);
if (dateTime.getMillisOfSecond() != 0) {
- res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss.SSS+08:00");
+ res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss.SSS" + Util.timeZone);
} else {
- res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss+08:00");
+ res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss" + Util.timeZone);
}
}
} else if (v.length() == 10 && v.charAt(4) == '-' && v.charAt(7) == '-') {
@@ -234,8 +231,11 @@ public class ESSyncUtil {
private static byte[] blobToBytes(Blob blob) {
try (InputStream is = blob.getBinaryStream()) {
byte[] b = new byte[(int) blob.length()];
- is.read(b);
- return b;
+ if (is.read(b) != -1) {
+ return b;
+ } else {
+ return new byte[0];
+ }
} catch (IOException | SQLException e) {
logger.error(e.getMessage());
return null;
@@ -297,46 +297,4 @@ public class ESSyncUtil {
sql.append(owner).append(".").append(columnName).append("=").append(value).append(" AND ");
}
}
-
- /**
- * 执行查询sql
- */
- public static Object sqlRS(DataSource ds, String sql, Function fun) {
- Connection conn = null;
- Statement smt = null;
- ResultSet rs = null;
- try {
- conn = ds.getConnection();
- smt = conn.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
- smt.setFetchSize(Integer.MIN_VALUE);
- rs = smt.executeQuery(sql);
-
- return fun.apply(rs);
- } catch (SQLException e) {
- logger.error("sqlRs has error, sql: {} ", sql);
- throw new RuntimeException(e);
- } finally {
- if (rs != null) {
- try {
- rs.close();
- } catch (SQLException e) {
- logger.error("error to close result set");
- }
- }
- if (smt != null) {
- try {
- smt.close();
- } catch (SQLException e) {
- logger.error("error to close statement");
- }
- }
- if (conn != null) {
- try {
- conn.close();
- } catch (SQLException e) {
- logger.error("error to close db connection");
- }
- }
- }
- }
}
diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESTemplate.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESTemplate.java
index 997f1366..8a083479 100644
--- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESTemplate.java
+++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESTemplate.java
@@ -2,6 +2,7 @@ package com.alibaba.otter.canal.client.adapter.es.support;
import java.sql.ResultSet;
import java.sql.SQLException;
+import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@@ -9,10 +10,13 @@ import java.util.concurrent.ConcurrentMap;
import javax.sql.DataSource;
+import org.apache.commons.lang.StringUtils;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
+import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.action.update.UpdateRequestBuilder;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.cluster.metadata.MappingMetaData;
import org.elasticsearch.common.collect.ImmutableOpenMap;
@@ -29,6 +33,7 @@ import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem;
import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem.ColumnItem;
import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem.FieldItem;
import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
+import com.alibaba.otter.canal.client.adapter.support.Util;
/**
* ES 操作模板
@@ -64,13 +69,24 @@ public class ESTemplate {
*/
public void insert(ESMapping mapping, Object pkVal, Map esFieldData) {
if (mapping.get_id() != null) {
+ String parentVal = (String) esFieldData.remove("$parent_routing");
if (mapping.isUpsert()) {
- getBulk().add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), pkVal.toString())
+ UpdateRequestBuilder updateRequestBuilder = transportClient
+ .prepareUpdate(mapping.get_index(), mapping.get_type(), pkVal.toString())
.setDoc(esFieldData)
- .setDocAsUpsert(true));
+ .setDocAsUpsert(true);
+ if (StringUtils.isNotEmpty(parentVal)) {
+ updateRequestBuilder.setRouting(parentVal);
+ }
+ getBulk().add(updateRequestBuilder);
} else {
- getBulk().add(transportClient.prepareIndex(mapping.get_index(), mapping.get_type(), pkVal.toString())
- .setSource(esFieldData));
+ IndexRequestBuilder indexRequestBuilder = transportClient
+ .prepareIndex(mapping.get_index(), mapping.get_type(), pkVal.toString())
+ .setSource(esFieldData);
+ if (StringUtils.isNotEmpty(parentVal)) {
+ indexRequestBuilder.setRouting(parentVal);
+ }
+ getBulk().add(indexRequestBuilder);
}
commitBulk();
} else {
@@ -96,7 +112,9 @@ public class ESTemplate {
* @param esFieldData 数据Map
*/
public void update(ESMapping mapping, Object pkVal, Map esFieldData) {
- append4Update(mapping, pkVal, esFieldData);
+ Map esFieldDataTmp = new LinkedHashMap<>(esFieldData.size());
+ esFieldData.forEach((k, v) -> esFieldDataTmp.put(Util.cleanColumn(k), v));
+ append4Update(mapping, pkVal, esFieldDataTmp);
commitBulk();
}
@@ -122,7 +140,7 @@ public class ESTemplate {
(fieldName, value) -> sql.append("_v.").append(fieldName).append("=").append(value).append(" AND "));
int len = sql.length();
sql.delete(len - 4, len);
- Integer syncCount = (Integer) ESSyncUtil.sqlRS(ds, sql.toString(), rs -> {
+ Integer syncCount = (Integer) Util.sqlRS(ds, sql.toString(), rs -> {
int count = 0;
try {
while (rs.next()) {
@@ -137,7 +155,7 @@ public class ESTemplate {
return count;
});
if (logger.isTraceEnabled()) {
- logger.trace("Update ES by query effect {} records", syncCount);
+ logger.trace("Update ES by query affected {} records", syncCount);
}
}
@@ -200,13 +218,24 @@ public class ESTemplate {
private void append4Update(ESMapping mapping, Object pkVal, Map esFieldData) {
if (mapping.get_id() != null) {
+ String parentVal = (String) esFieldData.remove("$parent_routing");
if (mapping.isUpsert()) {
- getBulk().add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), pkVal.toString())
+ UpdateRequestBuilder updateRequestBuilder = transportClient
+ .prepareUpdate(mapping.get_index(), mapping.get_type(), pkVal.toString())
.setDoc(esFieldData)
- .setDocAsUpsert(true));
+ .setDocAsUpsert(true);
+ if (StringUtils.isNotEmpty(parentVal)) {
+ updateRequestBuilder.setRouting(parentVal);
+ }
+ getBulk().add(updateRequestBuilder);
} else {
- getBulk().add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), pkVal.toString())
- .setDoc(esFieldData));
+ UpdateRequestBuilder updateRequestBuilder = transportClient
+ .prepareUpdate(mapping.get_index(), mapping.get_type(), pkVal.toString())
+ .setDoc(esFieldData);
+ if (StringUtils.isNotEmpty(parentVal)) {
+ updateRequestBuilder.setRouting(parentVal);
+ }
+ getBulk().add(updateRequestBuilder);
}
} else {
SearchResponse response = transportClient.prepareSearch(mapping.get_index())
@@ -223,6 +252,8 @@ public class ESTemplate {
public Object getValFromRS(ESMapping mapping, ResultSet resultSet, String fieldName,
String columnName) throws SQLException {
+ fieldName = Util.cleanColumn(fieldName);
+ columnName = Util.cleanColumn(columnName);
String esType = getEsType(mapping, fieldName);
Object value = resultSet.getObject(columnName);
@@ -254,9 +285,13 @@ public class ESTemplate {
if (!fieldItem.getFieldName().equals(mapping.get_id())
&& !mapping.getSkips().contains(fieldItem.getFieldName())) {
- esFieldData.put(fieldItem.getFieldName(), value);
+ esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value);
}
}
+
+ // 添加父子文档关联信息
+ putRelationDataFromRS(mapping, schemaItem, resultSet, esFieldData);
+
return resultIdVal;
}
@@ -288,12 +323,16 @@ public class ESTemplate {
for (ColumnItem columnItem : fieldItem.getColumnItems()) {
if (dmlOld.containsKey(columnItem.getColumnName())
&& !mapping.getSkips().contains(fieldItem.getFieldName())) {
- esFieldData.put(fieldItem.getFieldName(),
+ esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()),
getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName()));
break;
}
}
}
+
+ // 添加父子文档关联信息
+ putRelationDataFromRS(mapping, schemaItem, resultSet, esFieldData);
+
return resultIdVal;
}
@@ -337,9 +376,12 @@ public class ESTemplate {
if (!fieldItem.getFieldName().equals(mapping.get_id())
&& !mapping.getSkips().contains(fieldItem.getFieldName())) {
- esFieldData.put(fieldItem.getFieldName(), value);
+ esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value);
}
}
+
+ // 添加父子文档关联信息
+ putRelationData(mapping, schemaItem, dmlData, esFieldData);
return resultIdVal;
}
@@ -364,13 +406,67 @@ public class ESTemplate {
}
if (dmlOld.containsKey(columnName) && !mapping.getSkips().contains(fieldItem.getFieldName())) {
- esFieldData.put(fieldItem.getFieldName(),
+ esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()),
getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName));
}
}
+
+ // 添加父子文档关联信息
+ putRelationData(mapping, schemaItem, dmlOld, esFieldData);
return resultIdVal;
}
+ private void putRelationDataFromRS(ESMapping mapping, SchemaItem schemaItem, ResultSet resultSet,
+ Map esFieldData) {
+ // 添加父子文档关联信息
+ if (!mapping.getRelations().isEmpty()) {
+ mapping.getRelations().forEach((relationField, relationMapping) -> {
+ Map relations = new HashMap<>();
+ relations.put("name", relationMapping.getName());
+ if (StringUtils.isNotEmpty(relationMapping.getParent())) {
+ FieldItem parentFieldItem = schemaItem.getSelectFields().get(relationMapping.getParent());
+ Object parentVal;
+ try {
+ parentVal = getValFromRS(mapping,
+ resultSet,
+ parentFieldItem.getFieldName(),
+ parentFieldItem.getFieldName());
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ if (parentVal != null) {
+ relations.put("parent", parentVal.toString());
+ esFieldData.put("$parent_routing", parentVal.toString());
+
+ }
+ }
+ esFieldData.put(relationField, relations);
+ });
+ }
+ }
+
+ private void putRelationData(ESMapping mapping, SchemaItem schemaItem, Map dmlData,
+ Map esFieldData) {
+ // 添加父子文档关联信息
+ if (!mapping.getRelations().isEmpty()) {
+ mapping.getRelations().forEach((relationField, relationMapping) -> {
+ Map relations = new HashMap<>();
+ relations.put("name", relationMapping.getName());
+ if (StringUtils.isNotEmpty(relationMapping.getParent())) {
+ FieldItem parentFieldItem = schemaItem.getSelectFields().get(relationMapping.getParent());
+ String columnName = parentFieldItem.getColumnItems().iterator().next().getColumnName();
+ Object parentVal = getValFromData(mapping, dmlData, parentFieldItem.getFieldName(), columnName);
+ if (parentVal != null) {
+ relations.put("parent", parentVal.toString());
+ esFieldData.put("$parent_routing", parentVal.toString());
+
+ }
+ }
+ esFieldData.put(relationField, relations);
+ });
+ }
+ }
+
/**
* es 字段类型本地缓存
*/
diff --git a/client-adapter/elasticsearch/src/main/resources/es/biz_order.yml b/client-adapter/elasticsearch/src/main/resources/es/biz_order.yml
new file mode 100644
index 00000000..c49806fa
--- /dev/null
+++ b/client-adapter/elasticsearch/src/main/resources/es/biz_order.yml
@@ -0,0 +1,21 @@
+dataSourceKey: defaultDS
+destination: example
+groupId: g1
+esMapping:
+ _index: customer
+ _type: _doc
+ _id: _id
+ relations:
+ customer_order:
+ name: order
+ parent: customer_id
+ sql: "select concat('oid_', t.id) as _id,
+ t.customer_id,
+ t.id as order_id,
+ t.serial_code as order_serial,
+ t.c_time as order_time
+ from biz_order t"
+ skips:
+ - customer_id
+ etlCondition: "where t.c_time>='{0}'"
+ commitBatch: 3000
diff --git a/client-adapter/elasticsearch/src/main/resources/es/customer.yml b/client-adapter/elasticsearch/src/main/resources/es/customer.yml
new file mode 100644
index 00000000..06793a58
--- /dev/null
+++ b/client-adapter/elasticsearch/src/main/resources/es/customer.yml
@@ -0,0 +1,47 @@
+dataSourceKey: defaultDS
+destination: example
+groupId: g1
+esMapping:
+ _index: customer
+ _type: _doc
+ _id: id
+ relations:
+ customer_order:
+ name: customer
+ sql: "select t.id, t.name, t.email from customer t"
+ etlCondition: "where t.c_time>='{0}'"
+ commitBatch: 3000
+
+
+#{
+# "mappings":{
+# "_doc":{
+# "properties":{
+# "id": {
+# "type": "long"
+# },
+# "name": {
+# "type": "text"
+# },
+# "email": {
+# "type": "text"
+# },
+# "order_id": {
+# "type": "long"
+# },
+# "order_serial": {
+# "type": "text"
+# },
+# "order_time": {
+# "type": "date"
+# },
+# "customer_order":{
+# "type":"join",
+# "relations":{
+# "customer":"order"
+# }
+# }
+# }
+# }
+# }
+#}
diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/ConfigLoadTest.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/ConfigLoadTest.java
index a66acbb9..76d4a5a0 100644
--- a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/ConfigLoadTest.java
+++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/ConfigLoadTest.java
@@ -23,6 +23,7 @@ public class ConfigLoadTest {
public void testLoad() {
Map configMap = ESSyncConfigLoader.load(null);
ESSyncConfig config = configMap.get("mytest_user.yml");
+ config.validate();
Assert.assertNotNull(config);
Assert.assertEquals("defaultDS", config.getDataSourceKey());
ESSyncConfig.ESMapping esMapping = config.getEsMapping();
diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/ESTest.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/ESTest.java
new file mode 100644
index 00000000..143449c6
--- /dev/null
+++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/ESTest.java
@@ -0,0 +1,118 @@
+package com.alibaba.otter.canal.client.adapter.es.test;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import org.elasticsearch.action.bulk.BulkItemResponse;
+import org.elasticsearch.action.bulk.BulkRequestBuilder;
+import org.elasticsearch.action.bulk.BulkResponse;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.client.transport.TransportClient;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.transport.TransportAddress;
+import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.rest.RestStatus;
+import org.elasticsearch.search.SearchHit;
+import org.elasticsearch.transport.client.PreBuiltTransportClient;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class ESTest {
+
+ private TransportClient transportClient;
+
+ @Before
+ public void init() throws UnknownHostException {
+ Settings.Builder settingBuilder = Settings.builder();
+ settingBuilder.put("cluster.name", TestConstant.clusterName);
+ Settings settings = settingBuilder.build();
+ transportClient = new PreBuiltTransportClient(settings);
+ String[] hostArray = TestConstant.esHosts.split(",");
+ for (String host : hostArray) {
+ int i = host.indexOf(":");
+ transportClient.addTransportAddress(new TransportAddress(InetAddress.getByName(host.substring(0, i)),
+ Integer.parseInt(host.substring(i + 1))));
+ }
+ }
+
+ @Test
+ public void test01() {
+ SearchResponse response = transportClient.prepareSearch("test")
+ .setTypes("osm")
+ .setQuery(QueryBuilders.termQuery("_id", "1"))
+ .setSize(10000)
+ .get();
+ for (SearchHit hit : response.getHits()) {
+ System.out.println(hit.getSourceAsMap().get("data").getClass());
+ }
+ }
+
+ @Test
+ public void test02() {
+ Map esFieldData = new LinkedHashMap<>();
+ esFieldData.put("userId", 2L);
+ esFieldData.put("eventId", 4L);
+ esFieldData.put("eventName", "网络异常");
+ esFieldData.put("description", "第四个事件信息");
+
+ Map relations = new LinkedHashMap<>();
+ esFieldData.put("user_event", relations);
+ relations.put("name", "event");
+ relations.put("parent", "2");
+
+ BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk();
+ bulkRequestBuilder
+ .add(transportClient.prepareIndex("test", "osm", "2_4").setRouting("2").setSource(esFieldData));
+ commit(bulkRequestBuilder);
+ }
+
+ @Test
+ public void test03() {
+ Map esFieldData = new LinkedHashMap<>();
+ esFieldData.put("userId", 2L);
+ esFieldData.put("eventName", "网络异常1");
+
+ Map relations = new LinkedHashMap<>();
+ esFieldData.put("user_event", relations);
+ relations.put("name", "event");
+ relations.put("parent", "2");
+
+ BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk();
+ bulkRequestBuilder.add(transportClient.prepareUpdate("test", "osm", "2_4").setRouting("2").setDoc(esFieldData));
+ commit(bulkRequestBuilder);
+ }
+
+ @Test
+ public void test04() {
+ BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk();
+ bulkRequestBuilder.add(transportClient.prepareDelete("test", "osm", "2_4"));
+ commit(bulkRequestBuilder);
+ }
+
+ private void commit(BulkRequestBuilder bulkRequestBuilder) {
+ if (bulkRequestBuilder.numberOfActions() > 0) {
+ BulkResponse response = bulkRequestBuilder.execute().actionGet();
+ if (response.hasFailures()) {
+ for (BulkItemResponse itemResponse : response.getItems()) {
+ if (!itemResponse.isFailed()) {
+ continue;
+ }
+
+ if (itemResponse.getFailure().getStatus() == RestStatus.NOT_FOUND) {
+ System.out.println(itemResponse.getFailureMessage());
+ } else {
+ System.out.println("ES bulk commit error" + itemResponse.getFailureMessage());
+ }
+ }
+ }
+ }
+ }
+
+ @After
+ public void after() {
+ transportClient.close();
+ }
+}
diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/TestConstant.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/TestConstant.java
index 46b0f7a9..ab2d58da 100644
--- a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/TestConstant.java
+++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/TestConstant.java
@@ -11,9 +11,9 @@ public class TestConstant {
public final static String jdbcPassword = "121212";
public final static String esHosts = "127.0.0.1:9300";
- public final static String clusterNmae = "elasticsearch";
+ public final static String clusterName = "elasticsearch";
- public static DruidDataSource dataSource;
+ public final static DruidDataSource dataSource;
static {
dataSource = new DruidDataSource();
diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/Common.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/Common.java
index 3f9c10d4..960b72b5 100644
--- a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/Common.java
+++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/Common.java
@@ -22,7 +22,7 @@ public class Common {
outerAdapterConfig.setName("es");
outerAdapterConfig.setHosts(TestConstant.esHosts);
Map properties = new HashMap<>();
- properties.put("cluster.name", TestConstant.clusterNmae);
+ properties.put("cluster.name", TestConstant.clusterName);
outerAdapterConfig.setProperties(properties);
ESAdapter esAdapter = new ESAdapter();
diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java
index c307aec9..66af5f54 100644
--- a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java
+++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java
@@ -96,7 +96,7 @@ public class HbaseAdapter implements OuterAdapter {
hbaseSyncService = new HbaseSyncService(hbaseTemplate);
configMonitor = new HbaseConfigMonitor();
- configMonitor.init(this);
+ configMonitor.init(this, envProperties);
} catch (Exception e) {
throw new RuntimeException(e);
}
diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/monitor/HbaseConfigMonitor.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/monitor/HbaseConfigMonitor.java
index 803bbe1c..ac96f213 100644
--- a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/monitor/HbaseConfigMonitor.java
+++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/monitor/HbaseConfigMonitor.java
@@ -3,6 +3,7 @@ package com.alibaba.otter.canal.client.adapter.hbase.monitor;
import java.io.File;
import java.util.HashMap;
import java.util.Map;
+import java.util.Properties;
import org.apache.commons.io.filefilter.FileFilterUtils;
import org.apache.commons.io.monitor.FileAlterationListenerAdaptor;
@@ -11,8 +12,8 @@ import org.apache.commons.io.monitor.FileAlterationObserver;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.yaml.snakeyaml.Yaml;
+import com.alibaba.otter.canal.client.adapter.config.YmlConfigBinder;
import com.alibaba.otter.canal.client.adapter.hbase.HbaseAdapter;
import com.alibaba.otter.canal.client.adapter.hbase.config.MappingConfig;
import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader;
@@ -26,10 +27,13 @@ public class HbaseConfigMonitor {
private HbaseAdapter hbaseAdapter;
+ private Properties envProperties;
+
private FileAlterationMonitor fileMonitor;
- public void init(HbaseAdapter hbaseAdapter) {
+ public void init(HbaseAdapter hbaseAdapter, Properties envProperties) {
this.hbaseAdapter = hbaseAdapter;
+ this.envProperties = envProperties;
File confDir = Util.getConfDirPath(adapterName);
try {
FileAlterationObserver observer = new FileAlterationObserver(confDir,
@@ -60,7 +64,11 @@ public class HbaseConfigMonitor {
try {
// 加载新增的配置文件
String configContent = MappingConfigsLoader.loadConfig(adapterName + File.separator + file.getName());
- MappingConfig config = new Yaml().loadAs(configContent, MappingConfig.class);
+ MappingConfig config = YmlConfigBinder
+ .bindYmlToObj(null, configContent, MappingConfig.class, null, envProperties);
+ if (config == null) {
+ return;
+ }
config.validate();
addConfigToCache(file, config);
@@ -83,7 +91,11 @@ public class HbaseConfigMonitor {
onFileDelete(file);
return;
}
- MappingConfig config = new Yaml().loadAs(configContent, MappingConfig.class);
+ MappingConfig config = YmlConfigBinder
+ .bindYmlToObj(null, configContent, MappingConfig.class, null, envProperties);
+ if (config == null) {
+ return;
+ }
config.validate();
if (hbaseAdapter.getHbaseMapping().containsKey(file.getName())) {
deleteConfigFromCache(file);
diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseEtlService.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseEtlService.java
index d5a2ef58..dbbec06e 100644
--- a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseEtlService.java
+++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseEtlService.java
@@ -1,397 +1,391 @@
-package com.alibaba.otter.canal.client.adapter.hbase.service;
-
-import java.sql.ResultSetMetaData;
-import java.sql.Timestamp;
-import java.text.SimpleDateFormat;
-import java.util.ArrayList;
-import java.util.Date;
-import java.util.List;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicLong;
-
-import javax.sql.DataSource;
-
-import org.apache.hadoop.hbase.util.Bytes;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.alibaba.otter.canal.client.adapter.hbase.config.MappingConfig;
-import com.alibaba.otter.canal.client.adapter.hbase.support.HRow;
-import com.alibaba.otter.canal.client.adapter.hbase.support.HbaseTemplate;
-import com.alibaba.otter.canal.client.adapter.hbase.support.PhType;
-import com.alibaba.otter.canal.client.adapter.hbase.support.PhTypeUtil;
-import com.alibaba.otter.canal.client.adapter.hbase.support.Type;
-import com.alibaba.otter.canal.client.adapter.hbase.support.TypeUtil;
-import com.alibaba.otter.canal.client.adapter.support.EtlResult;
-import com.alibaba.otter.canal.client.adapter.support.JdbcTypeUtil;
-import com.alibaba.otter.canal.client.adapter.support.Util;
-import com.google.common.base.Joiner;
-
-/**
- * HBase ETL 操作业务类
- *
- * @author rewerma @ 2018-10-20
- * @version 1.0.0
- */
-public class HbaseEtlService {
-
- private static Logger logger = LoggerFactory.getLogger(HbaseEtlService.class);
-
-
- /**
- * 建表
- *
- * @param hbaseTemplate
- * @param config
- */
- public static void createTable(HbaseTemplate hbaseTemplate, MappingConfig config) {
- try {
- // 判断hbase表是否存在,不存在则建表
- MappingConfig.HbaseMapping hbaseMapping = config.getHbaseMapping();
- if (!hbaseTemplate.tableExists(hbaseMapping.getHbaseTable())) {
- hbaseTemplate.createTable(hbaseMapping.getHbaseTable(), hbaseMapping.getFamily());
- }
- } catch (Exception e) {
- logger.error(e.getMessage(), e);
- throw new RuntimeException(e);
- }
- }
-
- /**
- * 导入数据
- *
- * @param ds 数据源
- * @param hbaseTemplate hbaseTemplate
- * @param config 配置
- * @param params 筛选条件
- * @return 导入结果
- */
- public static EtlResult importData(DataSource ds, HbaseTemplate hbaseTemplate, MappingConfig config,
- List params) {
- EtlResult etlResult = new EtlResult();
- AtomicLong successCount = new AtomicLong();
- List errMsg = new ArrayList<>();
- String hbaseTable = "";
- try {
- if (config == null) {
- logger.error("Config is null!");
- etlResult.setSucceeded(false);
- etlResult.setErrorMessage("Config is null!");
- return etlResult;
- }
- MappingConfig.HbaseMapping hbaseMapping = config.getHbaseMapping();
- hbaseTable = hbaseMapping.getHbaseTable();
-
- long start = System.currentTimeMillis();
-
- if (params != null && params.size() == 1 && "rebuild".equalsIgnoreCase(params.get(0))) {
- logger.info(hbaseMapping.getHbaseTable() + " rebuild is starting!");
- // 如果表存在则删除
- if (hbaseTemplate.tableExists(hbaseMapping.getHbaseTable())) {
- hbaseTemplate.disableTable(hbaseMapping.getHbaseTable());
- hbaseTemplate.deleteTable(hbaseMapping.getHbaseTable());
- }
- params = null;
- } else {
- logger.info(hbaseMapping.getHbaseTable() + " etl is starting!");
- }
- createTable(hbaseTemplate, config);
-
- // 拼接sql
- String sql = "SELECT * FROM " + config.getHbaseMapping().getDatabase() + "." + hbaseMapping.getTable();
-
- // 拼接条件
- if (params != null && params.size() == 1 && hbaseMapping.getEtlCondition() == null) {
- AtomicBoolean stExists = new AtomicBoolean(false);
- // 验证是否有SYS_TIME字段
- Util.sqlRS(ds, sql, rs -> {
- try {
- ResultSetMetaData rsmd = rs.getMetaData();
- int cnt = rsmd.getColumnCount();
- for (int i = 1; i <= cnt; i++) {
- String columnName = rsmd.getColumnName(i);
- if ("SYS_TIME".equalsIgnoreCase(columnName)) {
- stExists.set(true);
- break;
- }
- }
- } catch (Exception e) {
- // ignore
- }
- return null;
- });
- if (stExists.get()) {
- sql += " WHERE SYS_TIME >= '" + params.get(0) + "' ";
- }
- } else if (hbaseMapping.getEtlCondition() != null && params != null) {
- String etlCondition = hbaseMapping.getEtlCondition();
- int size = params.size();
- for (int i = 0; i < size; i++) {
- etlCondition = etlCondition.replace("{" + i + "}", params.get(i));
- }
-
- sql += " " + etlCondition;
- }
-
- // 获取总数
- String countSql = "SELECT COUNT(1) FROM ( " + sql + ") _CNT ";
- long cnt = (Long) Util.sqlRS(ds, countSql, rs -> {
- Long count = null;
- try {
- if (rs.next()) {
- count = ((Number) rs.getObject(1)).longValue();
- }
- } catch (Exception e) {
- logger.error(e.getMessage(), e);
- }
- return count == null ? 0 : count;
- });
-
- // 当大于1万条记录时开启多线程
- if (cnt >= 10000) {
- int threadCount = 3;
- long perThreadCnt = cnt / threadCount;
- ExecutorService executor = Executors.newFixedThreadPool(threadCount);
- List> futures = new ArrayList<>(threadCount);
- for (int i = 0; i < threadCount; i++) {
- long offset = i * perThreadCnt;
- Long size = null;
- if (i != threadCount - 1) {
- size = perThreadCnt;
- }
- String sqlFinal;
- if (size != null) {
- sqlFinal = sql + " LIMIT " + offset + "," + size;
- } else {
- sqlFinal = sql + " LIMIT " + offset + "," + cnt;
- }
- Future future = executor.submit(
- () -> executeSqlImport(ds, sqlFinal, hbaseMapping, hbaseTemplate, successCount, errMsg));
- futures.add(future);
- }
-
- for (Future future : futures) {
- future.get();
- }
-
- executor.shutdown();
- } else {
- executeSqlImport(ds, sql, hbaseMapping, hbaseTemplate, successCount, errMsg);
- }
-
- logger.info(hbaseMapping.getHbaseTable() + " etl completed in: "
- + (System.currentTimeMillis() - start) / 1000 + "s!");
-
- etlResult.setResultMessage("导入HBase表 " + hbaseMapping.getHbaseTable() + " 数据:" + successCount.get() + " 条");
- } catch (Exception e) {
- logger.error(e.getMessage(), e);
- errMsg.add(hbaseTable + " etl failed! ==>" + e.getMessage());
- }
-
- if (errMsg.isEmpty()) {
- etlResult.setSucceeded(true);
- } else {
- etlResult.setErrorMessage(Joiner.on("\n").join(errMsg));
- }
- return etlResult;
- }
-
- /**
- * 执行导入
- *
- * @param ds
- * @param sql
- * @param hbaseMapping
- * @param hbaseTemplate
- * @param successCount
- * @param errMsg
- * @return
- */
- private static boolean executeSqlImport(DataSource ds, String sql, MappingConfig.HbaseMapping hbaseMapping,
- HbaseTemplate hbaseTemplate, AtomicLong successCount, List errMsg) {
- try {
- Util.sqlRS(ds, sql, rs -> {
- int i = 1;
-
- try {
- boolean complete = false;
- List rows = new ArrayList<>();
- String[] rowKeyColumns = null;
- if (hbaseMapping.getRowKey() != null) {
- rowKeyColumns = hbaseMapping.getRowKey().trim().split(",");
- }
- while (rs.next()) {
- int cc = rs.getMetaData().getColumnCount();
- int[] jdbcTypes = new int[cc];
- Class>[] classes = new Class[cc];
- for (int j = 1; j <= cc; j++) {
- int jdbcType = rs.getMetaData().getColumnType(j);
- jdbcTypes[j - 1] = jdbcType;
- classes[j - 1] = JdbcTypeUtil.jdbcType2javaType(jdbcType);
- }
- HRow row = new HRow();
-
- if (rowKeyColumns != null) {
- // 取rowKey字段拼接
- StringBuilder rowKeyVale = new StringBuilder();
- for (String rowKeyColumnName : rowKeyColumns) {
- Object obj = rs.getObject(rowKeyColumnName);
- if (obj != null) {
- rowKeyVale.append(obj.toString());
- }
- rowKeyVale.append("|");
- }
- int len = rowKeyVale.length();
- if (len > 0) {
- rowKeyVale.delete(len - 1, len);
- }
- row.setRowKey(Bytes.toBytes(rowKeyVale.toString()));
- }
-
- for (int j = 1; j <= cc; j++) {
- String columnName = rs.getMetaData().getColumnName(j);
-
- Object val = JdbcTypeUtil.getRSData(rs, columnName, jdbcTypes[j - 1]);
- if (val == null) {
- continue;
- }
-
- MappingConfig.ColumnItem columnItem = hbaseMapping.getColumnItems().get(columnName);
- // 没有配置映射
- if (columnItem == null) {
- String family = hbaseMapping.getFamily();
- String qualifile = columnName;
- if (hbaseMapping.isUppercaseQualifier()) {
- qualifile = qualifile.toUpperCase();
- }
- if (MappingConfig.Mode.STRING == hbaseMapping.getMode()) {
- if (hbaseMapping.getRowKey() == null && j == 1) {
- row.setRowKey(Bytes.toBytes(val.toString()));
- } else {
- row.addCell(family, qualifile, Bytes.toBytes(val.toString()));
- }
- } else if (MappingConfig.Mode.NATIVE == hbaseMapping.getMode()) {
- Type type = Type.getType(classes[j - 1]);
- if (hbaseMapping.getRowKey() == null && j == 1) {
- row.setRowKey(TypeUtil.toBytes(val, type));
- } else {
- row.addCell(family, qualifile, TypeUtil.toBytes(val, type));
- }
- } else if (MappingConfig.Mode.PHOENIX == hbaseMapping.getMode()) {
- PhType phType = PhType.getType(classes[j - 1]);
- if (hbaseMapping.getRowKey() == null && j == 1) {
- row.setRowKey(PhTypeUtil.toBytes(val, phType));
- } else {
- row.addCell(family, qualifile, PhTypeUtil.toBytes(val, phType));
- }
- }
- } else {
- // 如果不需要类型转换
- if (columnItem.getType() == null || "".equals(columnItem.getType())) {
- if (val instanceof java.sql.Date) {
- SimpleDateFormat dateFmt = new SimpleDateFormat("yyyy-MM-dd");
- val = dateFmt.format((Date) val);
- } else if (val instanceof Timestamp) {
- SimpleDateFormat datetimeFmt = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
- val = datetimeFmt.format((Date) val);
- }
-
- byte[] valBytes = Bytes.toBytes(val.toString());
- if (columnItem.isRowKey()) {
- if (columnItem.getRowKeyLen() != null) {
- valBytes = Bytes.toBytes(limitLenNum(columnItem.getRowKeyLen(), val));
- row.setRowKey(valBytes);
- } else {
- row.setRowKey(valBytes);
- }
- } else {
- row.addCell(columnItem.getFamily(), columnItem.getQualifier(), valBytes);
- }
- } else {
- if (MappingConfig.Mode.STRING == hbaseMapping.getMode()) {
- byte[] valBytes = Bytes.toBytes(val.toString());
- if (columnItem.isRowKey()) {
- if (columnItem.getRowKeyLen() != null) {
- valBytes = Bytes.toBytes(limitLenNum(columnItem.getRowKeyLen(), val));
- }
- row.setRowKey(valBytes);
- } else {
- row.addCell(columnItem.getFamily(), columnItem.getQualifier(), valBytes);
- }
- } else if (MappingConfig.Mode.NATIVE == hbaseMapping.getMode()) {
- Type type = Type.getType(columnItem.getType());
- if (columnItem.isRowKey()) {
- if (columnItem.getRowKeyLen() != null) {
- String v = limitLenNum(columnItem.getRowKeyLen(), val);
- row.setRowKey(Bytes.toBytes(v));
- } else {
- row.setRowKey(TypeUtil.toBytes(val, type));
- }
- } else {
- row.addCell(columnItem.getFamily(),
- columnItem.getQualifier(),
- TypeUtil.toBytes(val, type));
- }
- } else if (MappingConfig.Mode.PHOENIX == hbaseMapping.getMode()) {
- PhType phType = PhType.getType(columnItem.getType());
- if (columnItem.isRowKey()) {
- row.setRowKey(PhTypeUtil.toBytes(val, phType));
- } else {
- row.addCell(columnItem.getFamily(),
- columnItem.getQualifier(),
- PhTypeUtil.toBytes(val, phType));
- }
- }
- }
- }
- }
-
- if (row.getRowKey() == null) throw new RuntimeException("RowKey 值为空");
-
- rows.add(row);
- complete = false;
- if (i % hbaseMapping.getCommitBatch() == 0 && !rows.isEmpty()) {
- hbaseTemplate.puts(hbaseMapping.getHbaseTable(), rows);
- rows.clear();
- complete = true;
- }
- i++;
- successCount.incrementAndGet();
- if (logger.isDebugEnabled()) {
- logger.debug("successful import count:" + successCount.get());
- }
- }
-
- if (!complete && !rows.isEmpty()) {
- hbaseTemplate.puts(hbaseMapping.getHbaseTable(), rows);
- }
-
- } catch (Exception e) {
- logger.error(hbaseMapping.getHbaseTable() + " etl failed! ==>" + e.getMessage(), e);
- errMsg.add(hbaseMapping.getHbaseTable() + " etl failed! ==>" + e.getMessage());
- // throw new RuntimeException(e);
- }
- return i;
- });
- return true;
- } catch (Exception e) {
- logger.error(e.getMessage(), e);
- return false;
- }
- }
-
- private static String limitLenNum(int len, Object val) {
- if (val == null) {
- return null;
- }
- if (val instanceof Number) {
- return String.format("%0" + len + "d", (Number) ((Number) val).longValue());
- } else if (val instanceof String) {
- return String.format("%0" + len + "d", Long.parseLong((String) val));
- }
- return null;
- }
-}
+package com.alibaba.otter.canal.client.adapter.hbase.service;
+
+import java.sql.ResultSetMetaData;
+import java.sql.Timestamp;
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+import javax.sql.DataSource;
+
+import org.apache.hadoop.hbase.util.Bytes;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.alibaba.otter.canal.client.adapter.hbase.config.MappingConfig;
+import com.alibaba.otter.canal.client.adapter.hbase.support.HRow;
+import com.alibaba.otter.canal.client.adapter.hbase.support.HbaseTemplate;
+import com.alibaba.otter.canal.client.adapter.hbase.support.PhType;
+import com.alibaba.otter.canal.client.adapter.hbase.support.PhTypeUtil;
+import com.alibaba.otter.canal.client.adapter.hbase.support.Type;
+import com.alibaba.otter.canal.client.adapter.hbase.support.TypeUtil;
+import com.alibaba.otter.canal.client.adapter.support.EtlResult;
+import com.alibaba.otter.canal.client.adapter.support.JdbcTypeUtil;
+import com.alibaba.otter.canal.client.adapter.support.Util;
+import com.google.common.base.Joiner;
+
+/**
+ * HBase ETL 操作业务类
+ *
+ * @author rewerma @ 2018-10-20
+ * @version 1.0.0
+ */
+public class HbaseEtlService {
+
+ private static Logger logger = LoggerFactory.getLogger(HbaseEtlService.class);
+
+ /**
+ * 建表
+ *
+ * @param hbaseTemplate
+ * @param config
+ */
+ public static void createTable(HbaseTemplate hbaseTemplate, MappingConfig config) {
+ try {
+ // 判断hbase表是否存在,不存在则建表
+ MappingConfig.HbaseMapping hbaseMapping = config.getHbaseMapping();
+ if (!hbaseTemplate.tableExists(hbaseMapping.getHbaseTable())) {
+ hbaseTemplate.createTable(hbaseMapping.getHbaseTable(), hbaseMapping.getFamily());
+ }
+ } catch (Exception e) {
+ logger.error(e.getMessage(), e);
+ throw new RuntimeException(e);
+ }
+ }
+
+ /**
+ * 导入数据
+ *
+ * @param ds 数据源
+ * @param hbaseTemplate hbaseTemplate
+ * @param config 配置
+ * @param params 筛选条件
+ * @return 导入结果
+ */
+ public static EtlResult importData(DataSource ds, HbaseTemplate hbaseTemplate, MappingConfig config,
+ List params) {
+ EtlResult etlResult = new EtlResult();
+ AtomicLong successCount = new AtomicLong();
+ List errMsg = new ArrayList<>();
+ String hbaseTable = "";
+ try {
+ if (config == null) {
+ logger.error("Config is null!");
+ etlResult.setSucceeded(false);
+ etlResult.setErrorMessage("Config is null!");
+ return etlResult;
+ }
+ MappingConfig.HbaseMapping hbaseMapping = config.getHbaseMapping();
+ hbaseTable = hbaseMapping.getHbaseTable();
+
+ long start = System.currentTimeMillis();
+
+ if (params != null && params.size() == 1 && "rebuild".equalsIgnoreCase(params.get(0))) {
+ logger.info(hbaseMapping.getHbaseTable() + " rebuild is starting!");
+ // 如果表存在则删除
+ if (hbaseTemplate.tableExists(hbaseMapping.getHbaseTable())) {
+ hbaseTemplate.disableTable(hbaseMapping.getHbaseTable());
+ hbaseTemplate.deleteTable(hbaseMapping.getHbaseTable());
+ }
+ params = null;
+ } else {
+ logger.info(hbaseMapping.getHbaseTable() + " etl is starting!");
+ }
+ createTable(hbaseTemplate, config);
+
+ // 拼接sql
+ String sql = "SELECT * FROM " + config.getHbaseMapping().getDatabase() + "." + hbaseMapping.getTable();
+
+ // 拼接条件
+ if (params != null && params.size() == 1 && hbaseMapping.getEtlCondition() == null) {
+ AtomicBoolean stExists = new AtomicBoolean(false);
+ // 验证是否有SYS_TIME字段
+ Util.sqlRS(ds, sql, rs -> {
+ try {
+ ResultSetMetaData rsmd = rs.getMetaData();
+ int cnt = rsmd.getColumnCount();
+ for (int i = 1; i <= cnt; i++) {
+ String columnName = rsmd.getColumnName(i);
+ if ("SYS_TIME".equalsIgnoreCase(columnName)) {
+ stExists.set(true);
+ break;
+ }
+ }
+ } catch (Exception e) {
+ // ignore
+ }
+ return null;
+ });
+ if (stExists.get()) {
+ sql += " WHERE SYS_TIME >= '" + params.get(0) + "' ";
+ }
+ } else if (hbaseMapping.getEtlCondition() != null && params != null) {
+ String etlCondition = hbaseMapping.getEtlCondition();
+ int size = params.size();
+ for (int i = 0; i < size; i++) {
+ etlCondition = etlCondition.replace("{" + i + "}", params.get(i));
+ }
+
+ sql += " " + etlCondition;
+ }
+
+ // 获取总数
+ String countSql = "SELECT COUNT(1) FROM ( " + sql + ") _CNT ";
+ long cnt = (Long) Util.sqlRS(ds, countSql, rs -> {
+ Long count = null;
+ try {
+ if (rs.next()) {
+ count = ((Number) rs.getObject(1)).longValue();
+ }
+ } catch (Exception e) {
+ logger.error(e.getMessage(), e);
+ }
+ return count == null ? 0 : count;
+ });
+
+ // 当大于1万条记录时开启多线程
+ if (cnt >= 10000) {
+ int threadCount = 3;
+ long perThreadCnt = cnt / threadCount;
+ ExecutorService executor = Util.newFixedThreadPool(threadCount, 5000L);
+ for (int i = 0; i < threadCount; i++) {
+ long offset = i * perThreadCnt;
+ Long size = null;
+ if (i != threadCount - 1) {
+ size = perThreadCnt;
+ }
+ String sqlFinal;
+ if (size != null) {
+ sqlFinal = sql + " LIMIT " + offset + "," + size;
+ } else {
+ sqlFinal = sql + " LIMIT " + offset + "," + cnt;
+ }
+ executor.submit(
+ () -> executeSqlImport(ds, sqlFinal, hbaseMapping, hbaseTemplate, successCount, errMsg));
+ }
+
+ executor.shutdown();
+ while (!executor.awaitTermination(3, TimeUnit.SECONDS)) {
+ // ignore
+ }
+ } else {
+ executeSqlImport(ds, sql, hbaseMapping, hbaseTemplate, successCount, errMsg);
+ }
+
+ logger.info(hbaseMapping.getHbaseTable() + " etl completed in: "
+ + (System.currentTimeMillis() - start) / 1000 + "s!");
+
+ etlResult.setResultMessage("导入HBase表 " + hbaseMapping.getHbaseTable() + " 数据:" + successCount.get() + " 条");
+ } catch (Exception e) {
+ logger.error(e.getMessage(), e);
+ errMsg.add(hbaseTable + " etl failed! ==>" + e.getMessage());
+ }
+
+ if (errMsg.isEmpty()) {
+ etlResult.setSucceeded(true);
+ } else {
+ etlResult.setErrorMessage(Joiner.on("\n").join(errMsg));
+ }
+ return etlResult;
+ }
+
+ /**
+ * 执行导入
+ *
+ * @param ds
+ * @param sql
+ * @param hbaseMapping
+ * @param hbaseTemplate
+ * @param successCount
+ * @param errMsg
+ * @return
+ */
+ private static boolean executeSqlImport(DataSource ds, String sql, MappingConfig.HbaseMapping hbaseMapping,
+ HbaseTemplate hbaseTemplate, AtomicLong successCount, List errMsg) {
+ try {
+ Util.sqlRS(ds, sql, rs -> {
+ int i = 1;
+
+ try {
+ boolean complete = false;
+ List rows = new ArrayList<>();
+ String[] rowKeyColumns = null;
+ if (hbaseMapping.getRowKey() != null) {
+ rowKeyColumns = hbaseMapping.getRowKey().trim().split(",");
+ }
+ while (rs.next()) {
+ int cc = rs.getMetaData().getColumnCount();
+ int[] jdbcTypes = new int[cc];
+ Class>[] classes = new Class[cc];
+ for (int j = 1; j <= cc; j++) {
+ int jdbcType = rs.getMetaData().getColumnType(j);
+ jdbcTypes[j - 1] = jdbcType;
+ classes[j - 1] = JdbcTypeUtil.jdbcType2javaType(jdbcType);
+ }
+ HRow row = new HRow();
+
+ if (rowKeyColumns != null) {
+ // 取rowKey字段拼接
+ StringBuilder rowKeyVale = new StringBuilder();
+ for (String rowKeyColumnName : rowKeyColumns) {
+ Object obj = rs.getObject(rowKeyColumnName);
+ if (obj != null) {
+ rowKeyVale.append(obj.toString());
+ }
+ rowKeyVale.append("|");
+ }
+ int len = rowKeyVale.length();
+ if (len > 0) {
+ rowKeyVale.delete(len - 1, len);
+ }
+ row.setRowKey(Bytes.toBytes(rowKeyVale.toString()));
+ }
+
+ for (int j = 1; j <= cc; j++) {
+ String columnName = rs.getMetaData().getColumnName(j);
+
+ Object val = JdbcTypeUtil.getRSData(rs, columnName, jdbcTypes[j - 1]);
+ if (val == null) {
+ continue;
+ }
+
+ MappingConfig.ColumnItem columnItem = hbaseMapping.getColumnItems().get(columnName);
+ // 没有配置映射
+ if (columnItem == null) {
+ String family = hbaseMapping.getFamily();
+ String qualifile = columnName;
+ if (hbaseMapping.isUppercaseQualifier()) {
+ qualifile = qualifile.toUpperCase();
+ }
+ if (MappingConfig.Mode.STRING == hbaseMapping.getMode()) {
+ if (hbaseMapping.getRowKey() == null && j == 1) {
+ row.setRowKey(Bytes.toBytes(val.toString()));
+ } else {
+ row.addCell(family, qualifile, Bytes.toBytes(val.toString()));
+ }
+ } else if (MappingConfig.Mode.NATIVE == hbaseMapping.getMode()) {
+ Type type = Type.getType(classes[j - 1]);
+ if (hbaseMapping.getRowKey() == null && j == 1) {
+ row.setRowKey(TypeUtil.toBytes(val, type));
+ } else {
+ row.addCell(family, qualifile, TypeUtil.toBytes(val, type));
+ }
+ } else if (MappingConfig.Mode.PHOENIX == hbaseMapping.getMode()) {
+ PhType phType = PhType.getType(classes[j - 1]);
+ if (hbaseMapping.getRowKey() == null && j == 1) {
+ row.setRowKey(PhTypeUtil.toBytes(val, phType));
+ } else {
+ row.addCell(family, qualifile, PhTypeUtil.toBytes(val, phType));
+ }
+ }
+ } else {
+ // 如果不需要类型转换
+ if (columnItem.getType() == null || "".equals(columnItem.getType())) {
+ if (val instanceof java.sql.Date) {
+ SimpleDateFormat dateFmt = new SimpleDateFormat("yyyy-MM-dd");
+ val = dateFmt.format((Date) val);
+ } else if (val instanceof Timestamp) {
+ SimpleDateFormat datetimeFmt = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+ val = datetimeFmt.format((Date) val);
+ }
+
+ byte[] valBytes = Bytes.toBytes(val.toString());
+ if (columnItem.isRowKey()) {
+ if (columnItem.getRowKeyLen() != null) {
+ valBytes = Bytes.toBytes(limitLenNum(columnItem.getRowKeyLen(), val));
+ row.setRowKey(valBytes);
+ } else {
+ row.setRowKey(valBytes);
+ }
+ } else {
+ row.addCell(columnItem.getFamily(), columnItem.getQualifier(), valBytes);
+ }
+ } else {
+ if (MappingConfig.Mode.STRING == hbaseMapping.getMode()) {
+ byte[] valBytes = Bytes.toBytes(val.toString());
+ if (columnItem.isRowKey()) {
+ if (columnItem.getRowKeyLen() != null) {
+ valBytes = Bytes.toBytes(limitLenNum(columnItem.getRowKeyLen(), val));
+ }
+ row.setRowKey(valBytes);
+ } else {
+ row.addCell(columnItem.getFamily(), columnItem.getQualifier(), valBytes);
+ }
+ } else if (MappingConfig.Mode.NATIVE == hbaseMapping.getMode()) {
+ Type type = Type.getType(columnItem.getType());
+ if (columnItem.isRowKey()) {
+ if (columnItem.getRowKeyLen() != null) {
+ String v = limitLenNum(columnItem.getRowKeyLen(), val);
+ row.setRowKey(Bytes.toBytes(v));
+ } else {
+ row.setRowKey(TypeUtil.toBytes(val, type));
+ }
+ } else {
+ row.addCell(columnItem.getFamily(),
+ columnItem.getQualifier(),
+ TypeUtil.toBytes(val, type));
+ }
+ } else if (MappingConfig.Mode.PHOENIX == hbaseMapping.getMode()) {
+ PhType phType = PhType.getType(columnItem.getType());
+ if (columnItem.isRowKey()) {
+ row.setRowKey(PhTypeUtil.toBytes(val, phType));
+ } else {
+ row.addCell(columnItem.getFamily(),
+ columnItem.getQualifier(),
+ PhTypeUtil.toBytes(val, phType));
+ }
+ }
+ }
+ }
+ }
+
+ if (row.getRowKey() == null) throw new RuntimeException("RowKey 值为空");
+
+ rows.add(row);
+ complete = false;
+ if (i % hbaseMapping.getCommitBatch() == 0 && !rows.isEmpty()) {
+ hbaseTemplate.puts(hbaseMapping.getHbaseTable(), rows);
+ rows.clear();
+ complete = true;
+ }
+ i++;
+ successCount.incrementAndGet();
+ if (logger.isDebugEnabled()) {
+ logger.debug("successful import count:" + successCount.get());
+ }
+ }
+
+ if (!complete && !rows.isEmpty()) {
+ hbaseTemplate.puts(hbaseMapping.getHbaseTable(), rows);
+ }
+
+ } catch (Exception e) {
+ logger.error(hbaseMapping.getHbaseTable() + " etl failed! ==>" + e.getMessage(), e);
+ errMsg.add(hbaseMapping.getHbaseTable() + " etl failed! ==>" + e.getMessage());
+ // throw new RuntimeException(e);
+ }
+ return i;
+ });
+ return true;
+ } catch (Exception e) {
+ logger.error(e.getMessage(), e);
+ return false;
+ }
+ }
+
+ private static String limitLenNum(int len, Object val) {
+ if (val == null) {
+ return null;
+ }
+ if (val instanceof Number) {
+ return String.format("%0" + len + "d", (Number) ((Number) val).longValue());
+ } else if (val instanceof String) {
+ return String.format("%0" + len + "d", Long.parseLong((String) val));
+ }
+ return null;
+ }
+}
diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseSyncService.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseSyncService.java
index cb6f1b61..3a849c9e 100644
--- a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseSyncService.java
+++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseSyncService.java
@@ -20,9 +20,9 @@ import com.alibaba.otter.canal.client.adapter.support.Dml;
*/
public class HbaseSyncService {
- private Logger logger = LoggerFactory.getLogger(this.getClass());
+ private static Logger logger = LoggerFactory.getLogger(HbaseSyncService.class);
- private HbaseTemplate hbaseTemplate; // HBase操作模板
+ private HbaseTemplate hbaseTemplate; // HBase操作模板
public HbaseSyncService(HbaseTemplate hbaseTemplate){
this.hbaseTemplate = hbaseTemplate;
@@ -140,7 +140,7 @@ public class HbaseSyncService {
Integer.parseInt((String) entry.getValue()));
bytes = Bytes.toBytes(v);
} catch (Exception e) {
- // ignore
+ logger.error(e.getMessage(), e);
}
}
}
diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/PhTypeUtil.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/PhTypeUtil.java
index 6ff684e4..fc0cc909 100644
--- a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/PhTypeUtil.java
+++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/PhTypeUtil.java
@@ -7,12 +7,10 @@ import java.math.RoundingMode;
import java.sql.Timestamp;
import java.util.Date;
-import com.alibaba.otter.canal.client.adapter.support.Util;
import org.apache.hadoop.hbase.util.Bytes;
-import org.joda.time.DateTime;
+import com.alibaba.otter.canal.client.adapter.support.Util;
import com.google.common.math.LongMath;
-import org.joda.time.DateTimeZone;
/**
* Phoenix类型转换工具类
diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/common/EtlLock.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/common/EtlLock.java
index 5fd5f3fe..51ebb369 100644
--- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/common/EtlLock.java
+++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/common/EtlLock.java
@@ -27,7 +27,7 @@ public class EtlLock {
private static final Map DISTRIBUTED_LOCK = new ConcurrentHashMap<>();
- private static Mode mode = Mode.LOCAL;
+ private Mode mode = Mode.LOCAL;
@Resource
private CuratorClient curatorClient;
diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/common/SyncSwitch.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/common/SyncSwitch.java
index c0e5d394..577e354a 100644
--- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/common/SyncSwitch.java
+++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/common/SyncSwitch.java
@@ -34,7 +34,7 @@ public class SyncSwitch {
private static final Map DISTRIBUTED_LOCK = new ConcurrentHashMap<>();
- private static Mode mode = Mode.LOCAL;
+ private Mode mode = Mode.LOCAL;
@Resource
private AdapterCanalConfig adapterCanalConfig;
@@ -165,20 +165,20 @@ public class SyncSwitch {
}
}
- public Boolean status(String destination) {
+ public boolean status(String destination) {
if (mode == Mode.LOCAL) {
BooleanMutex mutex = LOCAL_LOCK.get(destination);
if (mutex != null) {
return mutex.state();
} else {
- return null;
+ return false;
}
} else {
BooleanMutex mutex = DISTRIBUTED_LOCK.get(destination);
if (mutex != null) {
return mutex.state();
} else {
- return null;
+ return false;
}
}
}
diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java
index ff88560c..de512923 100644
--- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java
+++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java
@@ -3,7 +3,6 @@ package com.alibaba.otter.canal.adapter.launcher.loader;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
@@ -17,6 +16,7 @@ import com.alibaba.otter.canal.client.adapter.OuterAdapter;
import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig;
import com.alibaba.otter.canal.client.adapter.support.Dml;
import com.alibaba.otter.canal.client.adapter.support.MessageUtil;
+import com.alibaba.otter.canal.client.adapter.support.Util;
import com.alibaba.otter.canal.protocol.FlatMessage;
import com.alibaba.otter.canal.protocol.Message;
@@ -43,7 +43,7 @@ public abstract class AbstractCanalAdapterWorker {
public AbstractCanalAdapterWorker(List> canalOuterAdapters){
this.canalOuterAdapters = canalOuterAdapters;
- this.groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size());
+ this.groupInnerExecutorService = Util.newFixedThreadPool(canalOuterAdapters.size(), 5000L);
syncSwitch = (SyncSwitch) SpringContext.getBean(SyncSwitch.class);
}
diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java
index 72291ccb..c46dff58 100644
--- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java
+++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java
@@ -2,13 +2,13 @@ package com.alibaba.otter.canal.adapter.launcher.loader;
import java.util.List;
import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.apache.kafka.common.errors.WakeupException;
import com.alibaba.otter.canal.client.adapter.OuterAdapter;
import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig;
+import com.alibaba.otter.canal.client.adapter.support.Util;
import com.alibaba.otter.canal.client.kafka.KafkaCanalConnector;
/**
@@ -44,12 +44,12 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker {
protected void process() {
while (!running) {
try {
- Thread.sleep(1000);
+ Thread.sleep(500);
} catch (InterruptedException e) {
// ignore
}
}
- ExecutorService workerExecutor = Executors.newSingleThreadExecutor();
+ ExecutorService workerExecutor = Util.newSingleThreadExecutor(5000L);
int retry = canalClientConfig.getRetries() == null
|| canalClientConfig.getRetries() == 0 ? 1 : canalClientConfig.getRetries();
long timeout = canalClientConfig.getTimeout() == null ? 30000 : canalClientConfig.getTimeout(); // 默认超时30秒
@@ -63,8 +63,8 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker {
connector.subscribe();
logger.info("=============> Subscribe topic: {} succeed <=============", this.topic);
while (running) {
- Boolean status = syncSwitch.status(canalDestination);
- if (status != null && !status) {
+ boolean status = syncSwitch.status(canalDestination);
+ if (!status) {
connector.disconnect();
break;
}
@@ -85,6 +85,8 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker {
}
}
+ workerExecutor.shutdown();
+
try {
connector.unsubscribe();
} catch (WakeupException e) {
diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java
index 12d5a040..11c26f4a 100644
--- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java
+++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java
@@ -5,7 +5,7 @@ import java.net.SocketAddress;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
@@ -176,40 +176,32 @@ public class CanalAdapterLoader {
public void destroy() {
if (!canalWorkers.isEmpty()) {
ExecutorService stopExecutorService = Executors.newFixedThreadPool(canalWorkers.size());
- List> futures = new ArrayList<>();
for (CanalAdapterWorker canalAdapterWorker : canalWorkers.values()) {
- futures.add(stopExecutorService.submit(() -> {
- canalAdapterWorker.stop();
- return true;
- }));
+ stopExecutorService.execute(canalAdapterWorker::stop);
}
- futures.forEach(future -> {
- try {
- future.get();
- } catch (Exception e) {
+ stopExecutorService.shutdown();
+ try {
+ while (!stopExecutorService.awaitTermination(1, TimeUnit.SECONDS)) {
// ignore
}
- });
- stopExecutorService.shutdown();
+ } catch (InterruptedException e) {
+ // ignore
+ }
}
if (!canalMQWorker.isEmpty()) {
ExecutorService stopMQWorkerService = Executors.newFixedThreadPool(canalMQWorker.size());
- List> futures = new ArrayList<>();
for (AbstractCanalAdapterWorker canalAdapterMQWorker : canalMQWorker.values()) {
- futures.add(stopMQWorkerService.submit(() -> {
- canalAdapterMQWorker.stop();
- return true;
- }));
+ stopMQWorkerService.execute(canalAdapterMQWorker::stop);
}
- futures.forEach(future -> {
- try {
- future.get();
- } catch (Exception e) {
+ stopMQWorkerService.shutdown();
+ try {
+ while (!stopMQWorkerService.awaitTermination(1, TimeUnit.SECONDS)) {
// ignore
}
- });
- stopMQWorkerService.shutdown();
+ } catch (InterruptedException e) {
+ // ignore
+ }
}
logger.info("All canal adapters destroyed");
}
diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java
index 369e8965..5da53763 100644
--- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java
+++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java
@@ -2,12 +2,12 @@ package com.alibaba.otter.canal.adapter.launcher.loader;
import java.util.List;
import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
import org.apache.kafka.common.errors.WakeupException;
import com.alibaba.otter.canal.client.adapter.OuterAdapter;
import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig;
+import com.alibaba.otter.canal.client.adapter.support.Util;
import com.alibaba.otter.canal.client.rocketmq.RocketMQCanalConnector;
/**
@@ -49,8 +49,9 @@ public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker {
}
}
- ExecutorService workerExecutor = Executors.newSingleThreadExecutor();
- int retry = canalClientConfig.getRetries() == null || canalClientConfig.getRetries() == 0 ? 1 : canalClientConfig.getRetries();
+ ExecutorService workerExecutor = Util.newSingleThreadExecutor(5000L);
+ int retry = canalClientConfig.getRetries() == null
+ || canalClientConfig.getRetries() == 0 ? 1 : canalClientConfig.getRetries();
long timeout = canalClientConfig.getTimeout() == null ? 30000 : canalClientConfig.getTimeout(); // 默认超时30秒
while (running) {
@@ -62,8 +63,8 @@ public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker {
connector.subscribe();
logger.info("=============> Subscribe topic: {} succeed<=============", this.topic);
while (running) {
- Boolean status = syncSwitch.status(canalDestination);
- if (status != null && !status) {
+ boolean status = syncSwitch.status(canalDestination);
+ if (!status) {
connector.disconnect();
break;
}
@@ -84,6 +85,8 @@ public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker {
}
}
+ workerExecutor.shutdown();
+
try {
connector.unsubscribe();
} catch (WakeupException e) {
diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java
index 5ed1063f..8f8cd06b 100644
--- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java
+++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java
@@ -140,7 +140,7 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker {
}
}
- } catch (Exception e) {
+ } catch (Throwable e) {
logger.error("process error!", e);
} finally {
connector.disconnect();
diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/ApplicationConfigMonitor.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/ApplicationConfigMonitor.java
index 8d519184..f7f86f11 100644
--- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/ApplicationConfigMonitor.java
+++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/ApplicationConfigMonitor.java
@@ -1,7 +1,9 @@
package com.alibaba.otter.canal.adapter.launcher.monitor;
import java.io.File;
-import java.io.FileReader;
+import java.io.FileInputStream;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
import java.util.Map;
import javax.annotation.PostConstruct;
@@ -68,7 +70,7 @@ public class ApplicationConfigMonitor {
super.onFileChange(file);
try {
// 检查yml格式
- new Yaml().loadAs(new FileReader(file), Map.class);
+ new Yaml().loadAs(new InputStreamReader(new FileInputStream(file), StandardCharsets.UTF_8), Map.class);
canalAdapterService.destroy();
diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/DbRemoteConfigLoader.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/DbRemoteConfigLoader.java
index 400e0463..0f1f821d 100644
--- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/DbRemoteConfigLoader.java
+++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/DbRemoteConfigLoader.java
@@ -1,6 +1,8 @@
package com.alibaba.otter.canal.adapter.launcher.monitor.remote;
-import java.io.FileWriter;
+import java.io.FileOutputStream;
+import java.io.OutputStreamWriter;
+import java.nio.charset.StandardCharsets;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
@@ -35,7 +37,7 @@ public class DbRemoteConfigLoader implements RemoteConfigLoader {
private DruidDataSource dataSource;
- private static volatile long currentConfigTimestamp = 0;
+ private volatile long currentConfigTimestamp = 0;
private Map remoteAdapterConfigs = new MapMaker().makeMap();
private ScheduledExecutorService executor = Executors.newScheduledThreadPool(2,
@@ -115,7 +117,10 @@ public class DbRemoteConfigLoader implements RemoteConfigLoader {
* @param content 文件内容
*/
private void overrideLocalCanalConfig(String content) {
- try (FileWriter writer = new FileWriter(CommonUtils.getConfPath() + "application.yml")) {
+
+ try (OutputStreamWriter writer = new OutputStreamWriter(
+ new FileOutputStream(CommonUtils.getConfPath() + "application.yml"),
+ StandardCharsets.UTF_8)) {
writer.write(content);
writer.flush();
} catch (Exception e) {
diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/RemoteAdapterMonitorImpl.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/RemoteAdapterMonitorImpl.java
index 6b4c2b81..2f0db822 100644
--- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/RemoteAdapterMonitorImpl.java
+++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/RemoteAdapterMonitorImpl.java
@@ -1,11 +1,14 @@
package com.alibaba.otter.canal.adapter.launcher.monitor.remote;
-import com.alibaba.otter.canal.common.utils.CommonUtils;
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.OutputStreamWriter;
+import java.nio.charset.StandardCharsets;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.File;
-import java.io.FileWriter;
+import com.alibaba.otter.canal.common.utils.CommonUtils;
/**
* 远程配置监听器实现
@@ -35,7 +38,9 @@ public class RemoteAdapterMonitorImpl implements RemoteAdapterMonitor {
}
}
String name = configItem.getName();
- try (FileWriter writer = new FileWriter(confPath + category + "/" + configItem.getName())) {
+ try (OutputStreamWriter writer = new OutputStreamWriter(
+ new FileOutputStream(confPath + category + "/" + configItem.getName()),
+ StandardCharsets.UTF_8)) {
writer.write(configItem.getContent());
writer.flush();
logger.info("## Loaded remote adapter config: {}/{}", category, name);
diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/rest/CommonRest.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/rest/CommonRest.java
index 302206be..2045b822 100644
--- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/rest/CommonRest.java
+++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/rest/CommonRest.java
@@ -69,16 +69,16 @@ public class CommonRest {
}
try {
- Boolean oriSwitchStatus;
+ boolean oriSwitchStatus;
if (destination != null) {
oriSwitchStatus = syncSwitch.status(destination);
- if (oriSwitchStatus != null && oriSwitchStatus) {
+ if (oriSwitchStatus) {
syncSwitch.off(destination);
}
} else {
// task可能为destination,直接锁task
oriSwitchStatus = syncSwitch.status(task);
- if (oriSwitchStatus != null && oriSwitchStatus) {
+ if (oriSwitchStatus) {
syncSwitch.off(task);
}
}
@@ -89,9 +89,9 @@ public class CommonRest {
}
return adapter.etl(task, paramArray);
} finally {
- if (destination != null && oriSwitchStatus != null && oriSwitchStatus) {
+ if (destination != null && oriSwitchStatus) {
syncSwitch.on(destination);
- } else if (destination == null && oriSwitchStatus != null && oriSwitchStatus) {
+ } else if (destination == null && oriSwitchStatus) {
syncSwitch.on(task);
}
}
@@ -102,7 +102,7 @@ public class CommonRest {
/**
* ETL curl http://127.0.0.1:8081/etl/hbase/mytest_person2.yml -X POST
- *
+ *
* @param type 类型 hbase, es
* @param task 任务名对应配置文件名 mytest_person2.yml
* @param params etl where条件参数, 为空全部导入
@@ -129,7 +129,7 @@ public class CommonRest {
/**
* 统计总数 curl http://127.0.0.1:8081/count/hbase/mytest_person2.yml
- *
+ *
* @param type 类型 hbase, es
* @param task 任务名对应配置文件名 mytest_person2.yml
* @return
@@ -148,11 +148,11 @@ public class CommonRest {
Set destinations = adapterCanalConfig.DESTINATIONS;
for (String destination : destinations) {
Map resMap = new LinkedHashMap<>();
- Boolean status = syncSwitch.status(destination);
- String resStatus = "none";
- if (status != null && status) {
+ boolean status = syncSwitch.status(destination);
+ String resStatus;
+ if (status) {
resStatus = "on";
- } else if (status != null && !status) {
+ } else {
resStatus = "off";
}
resMap.put("destination", destination);
@@ -164,7 +164,7 @@ public class CommonRest {
/**
* 实例同步开关 curl http://127.0.0.1:8081/syncSwitch/example/off -X PUT
- *
+ *
* @param destination 实例名称
* @param status 开关状态: off on
* @return
@@ -189,17 +189,17 @@ public class CommonRest {
/**
* 获取实例开关状态 curl http://127.0.0.1:8081/syncSwitch/example
- *
+ *
* @param destination 实例名称
* @return
*/
@GetMapping("/syncSwitch/{destination}")
public Map etl(@PathVariable String destination) {
- Boolean status = syncSwitch.status(destination);
- String resStatus = "none";
- if (status != null && status) {
+ boolean status = syncSwitch.status(destination);
+ String resStatus;
+ if (status) {
resStatus = "on";
- } else if (status != null && !status) {
+ } else {
resStatus = "off";
}
Map res = new LinkedHashMap<>();
diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java
index 8d1cab75..b45ad4d9 100644
--- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java
+++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java
@@ -146,7 +146,7 @@ public class RdbAdapter implements OuterAdapter {
skipDupException);
rdbConfigMonitor = new RdbConfigMonitor();
- rdbConfigMonitor.init(configuration.getKey(), this);
+ rdbConfigMonitor.init(configuration.getKey(), this, envProperties);
}
/**
diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfig.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfig.java
index d2421020..3381a1ca 100644
--- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfig.java
+++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfig.java
@@ -1,10 +1,10 @@
package com.alibaba.otter.canal.client.adapter.rdb.config;
-import org.apache.commons.lang.StringUtils;
-
import java.util.LinkedHashMap;
import java.util.Map;
+import org.apache.commons.lang.StringUtils;
+
/**
* RDB表映射配置
*
@@ -21,7 +21,7 @@ public class MappingConfig {
private String outerAdapterKey; // 对应适配器的key
- private Boolean concurrent; // 是否并行同步
+ private boolean concurrent = false; // 是否并行同步
private DbMapping dbMapping; // db映射配置
@@ -49,11 +49,11 @@ public class MappingConfig {
this.outerAdapterKey = outerAdapterKey;
}
- public Boolean getConcurrent() {
- return concurrent == null ? false : concurrent;
+ public boolean getConcurrent() {
+ return concurrent;
}
- public void setConcurrent(Boolean concurrent) {
+ public void setConcurrent(boolean concurrent) {
this.concurrent = concurrent;
}
@@ -87,11 +87,11 @@ public class MappingConfig {
public static class DbMapping {
- private Boolean mirrorDb = false; // 是否镜像库
+ private boolean mirrorDb = false; // 是否镜像库
private String database; // 数据库名或schema名
private String table; // 表名
private Map targetPk = new LinkedHashMap<>(); // 目标表主键字段
- private Boolean mapAll = false; // 映射所有字段
+ private boolean mapAll = false; // 映射所有字段
private String targetDb; // 目标库名
private String targetTable; // 目标表名
private Map targetColumns; // 目标表字段映射
@@ -103,11 +103,11 @@ public class MappingConfig {
private Map allMapColumns;
- public Boolean getMirrorDb() {
- return mirrorDb == null ? false : mirrorDb;
+ public boolean getMirrorDb() {
+ return mirrorDb;
}
- public void setMirrorDb(Boolean mirrorDb) {
+ public void setMirrorDb(boolean mirrorDb) {
this.mirrorDb = mirrorDb;
}
diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/monitor/RdbConfigMonitor.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/monitor/RdbConfigMonitor.java
index 779dc3b0..159748ba 100644
--- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/monitor/RdbConfigMonitor.java
+++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/monitor/RdbConfigMonitor.java
@@ -3,6 +3,7 @@ package com.alibaba.otter.canal.client.adapter.rdb.monitor;
import java.io.File;
import java.util.HashMap;
import java.util.Map;
+import java.util.Properties;
import org.apache.commons.io.filefilter.FileFilterUtils;
import org.apache.commons.io.monitor.FileAlterationListenerAdaptor;
@@ -11,8 +12,8 @@ import org.apache.commons.io.monitor.FileAlterationObserver;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.yaml.snakeyaml.Yaml;
+import com.alibaba.otter.canal.client.adapter.config.YmlConfigBinder;
import com.alibaba.otter.canal.client.adapter.rdb.RdbAdapter;
import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig;
import com.alibaba.otter.canal.client.adapter.rdb.config.MirrorDbConfig;
@@ -29,11 +30,14 @@ public class RdbConfigMonitor {
private RdbAdapter rdbAdapter;
+ private Properties envProperties;
+
private FileAlterationMonitor fileMonitor;
- public void init(String key, RdbAdapter rdbAdapter) {
+ public void init(String key, RdbAdapter rdbAdapter, Properties envProperties) {
this.key = key;
this.rdbAdapter = rdbAdapter;
+ this.envProperties = envProperties;
File confDir = Util.getConfDirPath(adapterName);
try {
FileAlterationObserver observer = new FileAlterationObserver(confDir,
@@ -64,7 +68,11 @@ public class RdbConfigMonitor {
try {
// 加载新增的配置文件
String configContent = MappingConfigsLoader.loadConfig(adapterName + File.separator + file.getName());
- MappingConfig config = new Yaml().loadAs(configContent, MappingConfig.class);
+ MappingConfig config = YmlConfigBinder
+ .bindYmlToObj(null, configContent, MappingConfig.class, null, envProperties);
+ if (config == null) {
+ return;
+ }
config.validate();
if ((key == null && config.getOuterAdapterKey() == null)
|| (key != null && key.equals(config.getOuterAdapterKey()))) {
@@ -90,7 +98,11 @@ public class RdbConfigMonitor {
onFileDelete(file);
return;
}
- MappingConfig config = new Yaml().loadAs(configContent, MappingConfig.class);
+ MappingConfig config = YmlConfigBinder
+ .bindYmlToObj(null, configContent, MappingConfig.class, null, envProperties);
+ if (config == null) {
+ return;
+ }
config.validate();
if ((key == null && config.getOuterAdapterKey() == null)
|| (key != null && key.equals(config.getOuterAdapterKey()))) {
diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java
index fbb188da..ed5f44e2 100644
--- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java
+++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java
@@ -6,8 +6,7 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
@@ -78,8 +77,7 @@ public class RdbEtlService {
if (cnt >= 10000) {
int threadCount = 3;
long perThreadCnt = cnt / threadCount;
- ExecutorService executor = Executors.newFixedThreadPool(threadCount);
- List> futures = new ArrayList<>(threadCount);
+ ExecutorService executor = Util.newFixedThreadPool(threadCount, 5000L);
for (int i = 0; i < threadCount; i++) {
long offset = i * perThreadCnt;
Long size = null;
@@ -92,16 +90,14 @@ public class RdbEtlService {
} else {
sqlFinal = sql + " LIMIT " + offset + "," + cnt;
}
- Future future = executor
- .submit(() -> executeSqlImport(srcDS, targetDS, sqlFinal, dbMapping, successCount, errMsg));
- futures.add(future);
- }
-
- for (Future future : futures) {
- future.get();
+ executor
+ .execute(() -> executeSqlImport(srcDS, targetDS, sqlFinal, dbMapping, successCount, errMsg));
}
executor.shutdown();
+ while (!executor.awaitTermination(3, TimeUnit.SECONDS)) {
+ // ignore
+ }
} else {
executeSqlImport(srcDS, targetDS, sql.toString(), dbMapping, successCount, errMsg);
}
@@ -202,7 +198,7 @@ public class RdbEtlService {
StringBuilder insertSql = new StringBuilder();
insertSql.append("INSERT INTO ").append(SyncUtil.getDbTableName(dbMapping)).append(" (");
columnsMap
- .forEach((targetColumnName, srcColumnName) -> insertSql.append(targetColumnName).append(","));
+ .forEach((targetColumnName, srcColumnName) -> insertSql.append(targetColumnName).append(","));
int len = insertSql.length();
insertSql.delete(len - 1, len).append(") VALUES (");
@@ -213,16 +209,18 @@ public class RdbEtlService {
len = insertSql.length();
insertSql.delete(len - 1, len).append(")");
try (Connection connTarget = targetDS.getConnection();
- PreparedStatement pstmt = connTarget.prepareStatement(insertSql.toString())) {
+ PreparedStatement pstmt = connTarget.prepareStatement(insertSql.toString())) {
connTarget.setAutoCommit(false);
while (rs.next()) {
+ completed = false;
+
pstmt.clearParameters();
// 删除数据
Map values = new LinkedHashMap<>();
StringBuilder deleteSql = new StringBuilder(
- "DELETE FROM " + SyncUtil.getDbTableName(dbMapping) + " WHERE ");
+ "DELETE FROM " + SyncUtil.getDbTableName(dbMapping) + " WHERE ");
appendCondition(dbMapping, deleteSql, values, rs);
try (PreparedStatement pstmt2 = connTarget.prepareStatement(deleteSql.toString())) {
int k = 1;
diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java
index 2c3ecfa9..a34bb784 100644
--- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java
+++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java
@@ -1,506 +1,507 @@
-package com.alibaba.otter.canal.client.adapter.rdb.service;
-
-import java.sql.Connection;
-import java.sql.ResultSetMetaData;
-import java.sql.SQLException;
-import java.util.ArrayList;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-import java.util.function.Function;
-
-import javax.sql.DataSource;
-
-import org.apache.commons.lang.StringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.alibaba.fastjson.JSON;
-import com.alibaba.fastjson.serializer.SerializerFeature;
-import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig;
-import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig.DbMapping;
-import com.alibaba.otter.canal.client.adapter.rdb.support.BatchExecutor;
-import com.alibaba.otter.canal.client.adapter.rdb.support.SingleDml;
-import com.alibaba.otter.canal.client.adapter.rdb.support.SyncUtil;
-import com.alibaba.otter.canal.client.adapter.support.Dml;
-import com.alibaba.otter.canal.client.adapter.support.Util;
-
-/**
- * RDB同步操作业务
- *
- * @author rewerma 2018-11-7 下午06:45:49
- * @version 1.0.0
- */
-public class RdbSyncService {
-
- private static final Logger logger = LoggerFactory.getLogger(RdbSyncService.class);
-
- // 源库表字段类型缓存: instance.schema.table ->
- private Map> columnsTypeCache;
-
- private int threads = 3;
- private boolean skipDupException;
-
- private List[] dmlsPartition;
- private BatchExecutor[] batchExecutors;
- private ExecutorService[] executorThreads;
-
- public List[] getDmlsPartition() {
- return dmlsPartition;
- }
-
- public Map> getColumnsTypeCache() {
- return columnsTypeCache;
- }
-
- public RdbSyncService(DataSource dataSource, Integer threads, boolean skipDupException){
- this(dataSource, threads, new ConcurrentHashMap<>(), skipDupException);
- }
-
- @SuppressWarnings("unchecked")
- public RdbSyncService(DataSource dataSource, Integer threads, Map> columnsTypeCache,
- boolean skipDupException){
- this.columnsTypeCache = columnsTypeCache;
- this.skipDupException = skipDupException;
- try {
- if (threads != null) {
- this.threads = threads;
- }
- this.dmlsPartition = new List[this.threads];
- this.batchExecutors = new BatchExecutor[this.threads];
- this.executorThreads = new ExecutorService[this.threads];
- for (int i = 0; i < this.threads; i++) {
- dmlsPartition[i] = new ArrayList<>();
- batchExecutors[i] = new BatchExecutor(dataSource);
- executorThreads[i] = Executors.newSingleThreadExecutor();
- }
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
-
- /**
- * 批量同步回调
- *
- * @param dmls 批量 DML
- * @param function 回调方法
- */
- public void sync(List dmls, Function function) {
- try {
- boolean toExecute = false;
- for (Dml dml : dmls) {
- if (!toExecute) {
- toExecute = function.apply(dml);
- } else {
- function.apply(dml);
- }
- }
- if (toExecute) {
- List> futures = new ArrayList<>();
- for (int i = 0; i < threads; i++) {
- int j = i;
- futures.add(executorThreads[i].submit(() -> {
- try {
- dmlsPartition[j]
- .forEach(syncItem -> sync(batchExecutors[j], syncItem.config, syncItem.singleDml));
- dmlsPartition[j].clear();
- batchExecutors[j].commit();
- return true;
- } catch (Throwable e) {
- batchExecutors[j].rollback();
- throw new RuntimeException(e);
- }
- }));
- }
-
- futures.forEach(future -> {
- try {
- future.get();
- } catch (ExecutionException | InterruptedException e) {
- throw new RuntimeException(e);
- }
- });
- }
- } finally {
- for (BatchExecutor batchExecutor : batchExecutors) {
- if (batchExecutor != null) {
- batchExecutor.close();
- }
- }
- }
- }
-
- /**
- * 批量同步
- *
- * @param mappingConfig 配置集合
- * @param dmls 批量 DML
- */
- public void sync(Map> mappingConfig, List dmls, Properties envProperties) {
- sync(dmls, dml -> {
- if (dml.getIsDdl() != null && dml.getIsDdl() && StringUtils.isNotEmpty(dml.getSql())) {
- // DDL
- columnsTypeCache.remove(dml.getDestination() + "." + dml.getDatabase() + "." + dml.getTable());
- return false;
- } else {
- // DML
- String destination = StringUtils.trimToEmpty(dml.getDestination());
- String groupId = StringUtils.trimToEmpty(dml.getGroupId());
- String database = dml.getDatabase();
- String table = dml.getTable();
- Map configMap;
- if (envProperties != null && !"tcp".equalsIgnoreCase(envProperties.getProperty("canal.conf.mode"))) {
- configMap = mappingConfig.get(destination + "-" + groupId + "_" + database + "-" + table);
- } else {
- configMap = mappingConfig.get(destination + "_" + database + "-" + table);
- }
-
- if (configMap == null) {
- return false;
- }
-
- if (configMap.values().isEmpty()) {
- return false;
- }
-
- for (MappingConfig config : configMap.values()) {
- if (config.getConcurrent()) {
- List singleDmls = SingleDml.dml2SingleDmls(dml);
- singleDmls.forEach(singleDml -> {
- int hash = pkHash(config.getDbMapping(), singleDml.getData());
- SyncItem syncItem = new SyncItem(config, singleDml);
- dmlsPartition[hash].add(syncItem);
- });
- } else {
- int hash = 0;
- List singleDmls = SingleDml.dml2SingleDmls(dml);
- singleDmls.forEach(singleDml -> {
- SyncItem syncItem = new SyncItem(config, singleDml);
- dmlsPartition[hash].add(syncItem);
- });
- }
- }
- return true;
- }
- });
- }
-
- /**
- * 单条 dml 同步
- *
- * @param batchExecutor 批量事务执行器
- * @param config 对应配置对象
- * @param dml DML
- */
- public void sync(BatchExecutor batchExecutor, MappingConfig config, SingleDml dml) {
- if (config != null) {
- try {
- String type = dml.getType();
- if (type != null && type.equalsIgnoreCase("INSERT")) {
- insert(batchExecutor, config, dml);
- } else if (type != null && type.equalsIgnoreCase("UPDATE")) {
- update(batchExecutor, config, dml);
- } else if (type != null && type.equalsIgnoreCase("DELETE")) {
- delete(batchExecutor, config, dml);
- } else if (type != null && type.equalsIgnoreCase("TRUNCATE")) {
- truncate(batchExecutor, config);
- }
- if (logger.isDebugEnabled()) {
- logger.debug("DML: {}", JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue));
- }
- } catch (SQLException e) {
- throw new RuntimeException(e);
- }
- }
- }
-
- /**
- * 插入操作
- *
- * @param config 配置项
- * @param dml DML数据
- */
- private void insert(BatchExecutor batchExecutor, MappingConfig config, SingleDml dml) throws SQLException {
- Map data = dml.getData();
- if (data == null || data.isEmpty()) {
- return;
- }
-
- DbMapping dbMapping = config.getDbMapping();
-
- Map columnsMap = SyncUtil.getColumnsMap(dbMapping, data);
-
- StringBuilder insertSql = new StringBuilder();
- insertSql.append("INSERT INTO ").append(SyncUtil.getDbTableName(dbMapping)).append(" (");
-
- columnsMap.forEach((targetColumnName, srcColumnName) -> insertSql.append(targetColumnName).append(","));
- int len = insertSql.length();
- insertSql.delete(len - 1, len).append(") VALUES (");
- int mapLen = columnsMap.size();
- for (int i = 0; i < mapLen; i++) {
- insertSql.append("?,");
- }
- len = insertSql.length();
- insertSql.delete(len - 1, len).append(")");
-
- Map ctype = getTargetColumnType(batchExecutor.getConn(), config);
-
- List