From 05ad671cc16f77e0980d63320212e78f7fff331d Mon Sep 17 00:00:00 2001 From: mcy Date: Wed, 13 Mar 2019 14:34:31 +0800 Subject: [PATCH 1/4] =?UTF-8?q?es=E5=A2=9E=E5=8A=A0relations=E9=85=8D?= =?UTF-8?q?=E7=BD=AE=20=E9=85=8D=E7=BD=AEmonitor=20yml=E5=8A=A0=E8=BD=BD?= =?UTF-8?q?=E4=BD=BF=E7=94=A8=E9=80=9A=E7=94=A8=E5=8A=A0=E8=BD=BD=E7=B1=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- client-adapter/elasticsearch/pom.xml | 6 + .../canal/client/adapter/es/ESAdapter.java | 2 +- .../adapter/es/config/ESSyncConfig.java | 77 +++++++++--- .../adapter/es/monitor/ESConfigMonitor.java | 25 ++-- .../adapter/es/test/ConfigLoadTest.java | 1 + .../canal/client/adapter/es/test/ESTest.java | 118 ++++++++++++++++++ .../client/adapter/es/test/TestConstant.java | 2 +- .../client/adapter/es/test/sync/Common.java | 2 +- .../hbase/monitor/HbaseConfigMonitor.java | 20 ++- .../canal/client/adapter/rdb/RdbAdapter.java | 2 +- .../adapter/rdb/monitor/RdbConfigMonitor.java | 20 ++- 11 files changed, 241 insertions(+), 34 deletions(-) create mode 100644 client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/ESTest.java 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..112f0287 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,7 +133,7 @@ public class ESAdapter implements OuterAdapter { esSyncService = new ESSyncService(esTemplate); esConfigMonitor = new ESConfigMonitor(); - esConfigMonitor.init(this); + esConfigMonitor.init(this, envProperties); } catch (Exception 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..a84223fe 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 @@ -5,6 +5,8 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import org.apache.commons.lang.StringUtils; + /** * ES 映射配置 * @@ -24,6 +26,12 @@ public class ESSyncConfig { private ESMapping esMapping; public void validate() { + if (!esMapping.relations.isEmpty()) { + Map relationMappings = esMapping.relations.get(0); + RelationMapping relationMapping = relationMappings.values().iterator().next(); + esMapping.isChild = StringUtils.isNotEmpty(relationMapping.getParent()); + } + if (esMapping._index == null) { throw new NullPointerException("esMapping._index"); } @@ -31,7 +39,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 +88,24 @@ 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 List> relations = new ArrayList<>(); + private boolean isChild = false; + 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 +163,22 @@ public class ESSyncConfig { this.skips = skips; } + public List> getRelations() { + return relations; + } + + public void setRelations(List> relations) { + this.relations = relations; + } + + public boolean isChild() { + return isChild; + } + + public void setChild(boolean child) { + isChild = child; + } + public String getSql() { return sql; } @@ -202,4 +227,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/monitor/ESConfigMonitor.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/monitor/ESConfigMonitor.java index 566a9fe0..8075f168 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,9 +3,11 @@ 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; +import com.alibaba.otter.canal.client.adapter.config.YmlConfigBinder; import org.apache.commons.io.filefilter.FileFilterUtils; import org.apache.commons.io.monitor.FileAlterationListenerAdaptor; import org.apache.commons.io.monitor.FileAlterationMonitor; @@ -31,10 +33,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 +70,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 +95,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/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..00f3e0de 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,7 +11,7 @@ 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; 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/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/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/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()))) { From 4bd91a10792b8d27f7807665718795bb551ad89b Mon Sep 17 00:00:00 2001 From: mcy Date: Wed, 13 Mar 2019 14:36:27 +0800 Subject: [PATCH 2/4] fix bug --- .../alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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); } From 4df60a0fb58313b4fcb7e817849324eb8a83de0d Mon Sep 17 00:00:00 2001 From: mcy Date: Wed, 13 Mar 2019 16:57:23 +0800 Subject: [PATCH 3/4] =?UTF-8?q?es=E5=A2=9E=E5=8A=A0=E7=88=B6=E5=AD=90?= =?UTF-8?q?=E6=96=87=E6=A1=A3=E7=B4=A2=E5=BC=95=E5=90=8C=E6=AD=A5=E9=80=82?= =?UTF-8?q?=E9=85=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../adapter/es/config/ESSyncConfig.java | 47 +++----- .../adapter/es/service/ESSyncService.java | 6 +- .../client/adapter/es/support/ESTemplate.java | 109 ++++++++++++++++-- .../src/main/resources/es/biz_order.yml | 21 ++++ .../src/main/resources/es/customer.yml | 47 ++++++++ 5 files changed, 187 insertions(+), 43 deletions(-) create mode 100644 client-adapter/elasticsearch/src/main/resources/es/biz_order.yml create mode 100644 client-adapter/elasticsearch/src/main/resources/es/customer.yml 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 a84223fe..da37189b 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 @@ -26,12 +26,6 @@ public class ESSyncConfig { private ESMapping esMapping; public void validate() { - if (!esMapping.relations.isEmpty()) { - Map relationMappings = esMapping.relations.get(0); - RelationMapping relationMapping = relationMappings.values().iterator().next(); - esMapping.isChild = StringUtils.isNotEmpty(relationMapping.getParent()); - } - if (esMapping._index == null) { throw new NullPointerException("esMapping._index"); } @@ -88,24 +82,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 List> relations = new ArrayList<>(); - private boolean isChild = false; - 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; @@ -163,22 +156,14 @@ public class ESSyncConfig { this.skips = skips; } - public List> getRelations() { + public Map getRelations() { return relations; } - public void setRelations(List> relations) { + public void setRelations(Map relations) { this.relations = relations; } - public boolean isChild() { - return isChild; - } - - public void setChild(boolean child) { - isChild = child; - } - public String getSql() { return sql; } 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..d5258223 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 @@ -43,7 +43,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 +65,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 +74,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()); } 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..0a67ff78 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; @@ -64,13 +68,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 { @@ -137,7 +152,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 +215,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()) @@ -257,6 +283,10 @@ public class ESTemplate { esFieldData.put(fieldItem.getFieldName(), value); } } + + // 添加父子文档关联信息 + putRelationDataFromRS(mapping, schemaItem, resultSet, esFieldData); + return resultIdVal; } @@ -294,6 +324,10 @@ public class ESTemplate { } } } + + // 添加父子文档关联信息 + putRelationDataFromRS(mapping, schemaItem, resultSet, esFieldData); + return resultIdVal; } @@ -340,6 +374,9 @@ public class ESTemplate { esFieldData.put(fieldItem.getFieldName(), value); } } + + // 添加父子文档关联信息 + putRelationData(mapping, schemaItem, dmlData, esFieldData); return resultIdVal; } @@ -368,9 +405,63 @@ public class ESTemplate { 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" +# } +# } +# } +# } +# } +#} From e03705e2112d0435f18183a777a333851d392a86 Mon Sep 17 00:00:00 2001 From: mcy Date: Wed, 13 Mar 2019 17:19:08 +0800 Subject: [PATCH 4/4] =?UTF-8?q?es=E7=88=B6=E5=AD=90=E6=96=87=E6=A1=A3?= =?UTF-8?q?=E7=B4=A2=E5=BC=95ETL=E6=94=AF=E6=8C=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../adapter/es/service/ESEtlService.java | 56 +++++++++++++++---- 1 file changed, 46 insertions(+), 10 deletions(-) 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..2a60dcd3 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,9 +1,7 @@ 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; @@ -13,13 +11,14 @@ 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; @@ -199,21 +198,58 @@ public class ESEtlService { Object val = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName); esFieldData.put(fieldName, val); } + + 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(relationField, relations); + }); + } + Object idVal = null; if (mapping.get_id() != null) { idVal = rs.getObject(mapping.get_id()); } 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());