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 ccd58773..f0505aaa 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 @@ -23,7 +23,7 @@ public class SchemaItem { this.isAllFieldsSimple(); aliasTableItems.values().forEach(tableItem -> { tableItem.getRelationTableFields(); - tableItem.getRelationKeyFieldItems(); + // tableItem.getRelationKeyFieldItems(); tableItem.getRelationSelectFieldItems(); }); } @@ -145,8 +145,9 @@ public class SchemaItem { private boolean main; private boolean subQuery; - private volatile Map> relationTableFields; // 当前表关联条件字段 - private volatile List relationKeyFieldItems; // 关联条件字段在select中的对应字段 + private volatile Map> relationTableFields; // 当前表关联条件字段对应主表查询字段 + // private volatile List relationKeyFieldItems; // + // 关联条件字段在select中的对应字段 private volatile List relationSelectFieldItems; // 子表所在主表的查询字段 public TableItem(SchemaItem schemaItem){ @@ -267,34 +268,36 @@ public class SchemaItem { return relationTableFields; } - public List getRelationKeyFieldItems() { - if (relationKeyFieldItems == null) { - synchronized (SchemaItem.class) { - if (relationKeyFieldItems == null) { - relationKeyFieldItems = new ArrayList<>(); - getRelationFields().forEach(relationFieldsPair -> { - FieldItem leftFieldItem = relationFieldsPair.getLeftFieldItem(); - List selectFieldItem = getSchemaItem().getColumnFields() - .get(leftFieldItem.getOwner() + "." + leftFieldItem.getColumn().getColumnName()); - if (selectFieldItem != null && !selectFieldItem.isEmpty()) { - relationKeyFieldItems.addAll(selectFieldItem); - } else { - FieldItem rightFieldItem = relationFieldsPair.getRightFieldItem(); - selectFieldItem = getSchemaItem().getColumnFields() - .get(rightFieldItem.getOwner() + "." + rightFieldItem.getColumn().getColumnName()); - if (selectFieldItem != null && !selectFieldItem.isEmpty()) { - relationKeyFieldItems.addAll(selectFieldItem); - } else { - throw new UnsupportedOperationException( - "Relation condition column must in select columns."); - } - } - }); - } - } - } - return relationKeyFieldItems; - } + // public List getRelationKeyFieldItems() { + // if (relationKeyFieldItems == null) { + // synchronized (SchemaItem.class) { + // if (relationKeyFieldItems == null) { + // relationKeyFieldItems = new ArrayList<>(); + // getRelationFields().forEach(relationFieldsPair -> { + // FieldItem leftFieldItem = relationFieldsPair.getLeftFieldItem(); + // List selectFieldItem = getSchemaItem().getColumnFields() + // .get(leftFieldItem.getOwner() + "." + + // leftFieldItem.getColumn().getColumnName()); + // if (selectFieldItem != null && !selectFieldItem.isEmpty()) { + // relationKeyFieldItems.addAll(selectFieldItem); + // } else { + // FieldItem rightFieldItem = relationFieldsPair.getRightFieldItem(); + // selectFieldItem = getSchemaItem().getColumnFields() + // .get(rightFieldItem.getOwner() + "." + + // rightFieldItem.getColumn().getColumnName()); + // if (selectFieldItem != null && !selectFieldItem.isEmpty()) { + // relationKeyFieldItems.addAll(selectFieldItem); + // } else { + // throw new UnsupportedOperationException( + // "Relation condition column must in select columns."); + // } + // } + // }); + // } + // } + // } + // return relationKeyFieldItems; + // } public List getRelationSelectFieldItems() { if (relationSelectFieldItems == null) { 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 9eb69c12..4e86383e 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 @@ -190,41 +190,46 @@ public class ESSyncService { if (allFieldsSimple) { // 不是子查询 if (!tableItem.isSubQuery()) { - Map esFieldData = new LinkedHashMap<>(); for (FieldItem fieldItem : tableItem.getRelationSelectFieldItems()) { Object value = esTemplate .getValFromData(mapping, data, fieldItem.getColumn().getColumnName()); esFieldData.put(fieldItem.getFieldName(), value); } + BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery(); for (Map.Entry> entry : tableItem.getRelationTableFields() .entrySet()) { Object value = esTemplate .getValFromData(mapping, data, entry.getKey().getColumn().getColumnName()); for (FieldItem fieldItem : entry.getValue()) { - queryBuilder.must(QueryBuilders.termsQuery(fieldItem.getFieldName(), value)); + String fieldName = fieldItem.getFieldName(); + // 判断是否是主键 + if (fieldName.equals(mapping.get_id())) { + fieldName = "_id"; + } + queryBuilder.must(QueryBuilders.termsQuery(fieldName, value)); } } - // - // for (FieldItem fieldItem : tableItem.getRelationTableFields()) { - // Object value = esTemplate - // .getValFromData(mapping, data, fieldItem.getColumn().getColumnName()); - // - // } - - // if (logger.isDebugEnabled()) { - // logger.debug("从表insert, 且均为简单字段，对es进行update_by_query, queryBuilder: {}, - // esFieldData:{}", queryBuilder.toString(), esFieldData); - // } + if (logger.isDebugEnabled()) { + logger.trace( + "Join table update es index by foreign key, destination:{}, table: {}, index: {}", + config.getDestination(), + dml.getTable(), + mapping.get_index()); + } boolean result = esTemplate.updateByQuery(mapping, queryBuilder, esFieldData); - // if (!result) { - // logger.error("从表insert，均为简单字段，直接对es进行update_by_query, es更新存在错误, table: {}, - // index: {}, dml: {}", dml.getTable(), config.getEsSyn().getIndex(), dml); - // } + if (!result) { + logger.error( + "Join table update es index by foreign key error, destination:{}, table: {}, index: {}", + config.getDestination(), + dml.getTable(), + mapping.get_index()); + } + } else { + // TODO } - // TODO } else { // TODO 查询总sql } 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 9908c0d8..fec06dd1 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 @@ -264,7 +264,7 @@ public class ESTemplate { if (itemResponse.getFailure().getStatus() == RestStatus.NOT_FOUND) { logger.warn(itemResponse.getFailureMessage()); } else { - logger.error("ES 同步数据错误 {}", itemResponse.getFailureMessage()); + logger.error("ES sync commit error: {}", itemResponse.getFailureMessage()); } } } @@ -314,7 +314,7 @@ public class ESTemplate { value = resultSet.getByte(fieldName); } } - return value; + return ESSyncUtil.typeConvert(value, esType); } public Object getESDataFromRS(ESMapping mapping, ResultSet resultSet, diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/SqlParseTest.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/SqlParseTest.java index f31e6a89..d07defe9 100644 --- a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/SqlParseTest.java +++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/SqlParseTest.java @@ -39,8 +39,9 @@ public class SqlParseTest { .forEach(fieldItem -> Assert.assertEquals("user_id", fieldItem.getColumn().getColumnName())); // 获取关联字段在select中的对应字段 - List relationSelectFieldItem = tableItem.getRelationKeyFieldItems(); - relationSelectFieldItem.forEach(fieldItem -> Assert.assertEquals("c.labels", - fieldItem.getOwner() + "." + fieldItem.getColumn().getColumnName())); + // List relationSelectFieldItem = + // tableItem.getRelationKeyFieldItems(); + // relationSelectFieldItem.forEach(fieldItem -> Assert.assertEquals("c.labels", + // fieldItem.getOwner() + "." + fieldItem.getColumn().getColumnName())); } } diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncTest.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncTest.java index 9fc9c211..c69fa995 100644 --- a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncTest.java +++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncTest.java @@ -3,6 +3,9 @@ package com.alibaba.otter.canal.client.adapter.es.test.sync; import java.util.*; import org.elasticsearch.action.get.GetResponse; +import org.elasticsearch.action.search.SearchResponse; +import org.elasticsearch.index.query.QueryBuilders; +import org.elasticsearch.search.SearchHit; import org.junit.After; import org.junit.Assert; import org.junit.Before; @@ -53,6 +56,7 @@ public class UserSyncTest { dataList.add(data); data.put("id", 1L); data.put("name", "Eric"); + data.put("role_id", 1L); data.put("c_time", new Date()); dml.setData(dataList); @@ -79,6 +83,7 @@ public class UserSyncTest { dataList.add(data); data.put("id", 1L); data.put("name", "Eric"); + data.put("role_id", 1L); data.put("c_time", new Date()); dml.setData(dataList); @@ -108,6 +113,18 @@ public class UserSyncTest { esAdapter.getEsSyncService().sync(dml); } + @Test + public void ttt(){ + SearchResponse response = esAdapter.getTransportClient().prepareSearch("mytest_user") + .setTypes("_doc") + .setQuery(QueryBuilders.termQuery("_id", 1L)) + .setSize(100) + .get(); + for (SearchHit hit : response.getHits()) { + System.out.println(hit.getSourceAsMap().get("name")); + } + } + @After public void after() { esAdapter.destroy(); diff --git a/client-adapter/elasticsearch/src/test/resources/log4j2-test.xml b/client-adapter/elasticsearch/src/test/resources/log4j2-test.xml new file mode 100644 index 00000000..9c3d2ee6 --- /dev/null +++ b/client-adapter/elasticsearch/src/test/resources/log4j2-test.xml @@ -0,0 +1,13 @@ + + + + + + + + + + + + + \ No newline at end of file diff --git a/client-adapter/launcher/src/main/resources/application.yml b/client-adapter/launcher/src/main/resources/application.yml index a7d512c4..3e0e0dc4 100644 --- a/client-adapter/launcher/src/main/resources/application.yml +++ b/client-adapter/launcher/src/main/resources/application.yml @@ -16,34 +16,34 @@ hbase.zookeeper.property.clientPort: 2181 hbase.zookeeper.znode.parent: /hbase canal.conf: -# canalServerHost: 127.0.0.1:11111 + canalServerHost: 127.0.0.1:11111 # zookeeperHosts: slave1:2181 - bootstrapServers: slave1:6667 #or rocketmq nameservers:host1:9876;host2:9876 +# bootstrapServers: slave1:6667 #or rocketmq nameservers:host1:9876;host2:9876 flatMessage: true retry: 3 timeout: 20000 batchSize: 50 -# canalInstances: -# - instance: example -# groups: -# - outAdapters: -# - name: es -# hosts: 127.0.0.1:9300 -# properties: -# cluster.name: elasticsearch + canalInstances: + - instance: example + groups: + - outAdapters: + - name: es + hosts: 127.0.0.1:9300 + properties: + cluster.name: elasticsearch # - name: logger # - name: hbase # properties: # hbase.zookeeper.quorum: ${hbase.zookeeper.quorum} # hbase.zookeeper.property.clientPort: ${hbase.zookeeper.property.clientPort} # zookeeper.znode.parent: ${hbase.zookeeper.znode.parent} - mqTopics: - - mqMode: kafka - topic: example - groups: - - groupId: g2 - outAdapters: - - name: logger +# mqTopics: +# - mqMode: kafka +# topic: example +# groups: +# - groupId: g2 +# outAdapters: +# - name: logger # mqTopics: # - mqMode: rocketmq # topic: example