From 1d2e85672ca85dda301ab04c02d6a93654d7f5c0 Mon Sep 17 00:00:00 2001 From: mcy Date: Tue, 30 Oct 2018 09:49:04 +0800 Subject: [PATCH] =?UTF-8?q?elasticsearch=E9=80=82=E9=85=8D=E5=99=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- client-adapter/elasticsearch/pom.xml | 77 ++++ .../canal/client/adapter/es/ESAdapter.java | 82 ++++ .../adapter/es/config/ESSyncConfig.java | 137 +++++++ .../adapter/es/config/ESSyncConfigLoader.java | 89 +++++ .../client/adapter/es/config/SchemaItem.java | 364 ++++++++++++++++++ .../client/adapter/es/config/SqlParser.java | 205 ++++++++++ .../adapter/es/service/ESSyncService.java | 15 + ...ba.otter.canal.client.adapter.OuterAdapter | 1 + .../src/main/resources/es/myetst_user.yml | 12 + .../adapter/es/test/ConfigLoadTest.java | 31 ++ .../client/adapter/es/test/SqlParseTest.java | 51 +++ client-adapter/pom.xml | 1 + 12 files changed, 1065 insertions(+) create mode 100644 client-adapter/elasticsearch/pom.xml create mode 100644 client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java create mode 100644 client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfig.java create mode 100644 client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfigLoader.java create mode 100644 client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SchemaItem.java create mode 100644 client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SqlParser.java create mode 100644 client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java create mode 100644 client-adapter/elasticsearch/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.OuterAdapter create mode 100644 client-adapter/elasticsearch/src/main/resources/es/myetst_user.yml create mode 100644 client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/ConfigLoadTest.java create mode 100644 client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/SqlParseTest.java diff --git a/client-adapter/elasticsearch/pom.xml b/client-adapter/elasticsearch/pom.xml new file mode 100644 index 00000000..b5a8c11b --- /dev/null +++ b/client-adapter/elasticsearch/pom.xml @@ -0,0 +1,77 @@ + + + + canal.client-adapter + com.alibaba.otter + 1.1.1-SNAPSHOT + + 4.0.0 + com.alibaba.otter + client-adapter.elasticsearch + jar + canal client adapter ElasticSearch module for otter ${project.version} + + + + com.alibaba.otter + client-adapter.common + ${project.version} + provided + + + org.yaml + snakeyaml + 1.19 + provided + + + com.alibaba.fastsql + fastsql + 2.0.0_preview_644 + + + org.elasticsearch + elasticsearch + 6.2.3 + + + org.elasticsearch.client + transport + 6.2.3 + + + + junit + junit + 4.12 + test + + + + + + + org.apache.maven.plugins + maven-assembly-plugin + 2.4 + + + jar-with-dependencies + + + + + make-assembly + package + + single + + + + + + + + \ No newline at end of file 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 new file mode 100644 index 00000000..8e2523e4 --- /dev/null +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java @@ -0,0 +1,82 @@ +package com.alibaba.otter.canal.client.adapter.es; + +import java.util.List; +import java.util.Map; + +import com.alibaba.otter.canal.client.adapter.es.service.ESSyncService; +import org.elasticsearch.client.transport.TransportClient; +import org.elasticsearch.common.settings.Settings; +import org.elasticsearch.transport.client.PreBuiltTransportClient; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.alibaba.otter.canal.client.adapter.OuterAdapter; +import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig; +import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfigLoader; +import com.alibaba.otter.canal.client.adapter.support.Dml; +import com.alibaba.otter.canal.client.adapter.support.EtlResult; +import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig; +import com.alibaba.otter.canal.client.adapter.support.SPI; + +/** + * ES外部适配器 + * + * @author rewerma 2018-10-20 + * @version 1.0.0 + */ +@SPI("es") +public class ESAdapter implements OuterAdapter { + + private static Logger logger = LoggerFactory.getLogger(ESAdapter.class); + + private static volatile Map esSyncConfig = null; // 文件名对应配置 + + private TransportClient transportClient; + + private ESSyncService esSyncService; + + @Override + public void init(OuterAdapterConfig configuration) { + try { + if (esSyncConfig == null) { + synchronized (ESSyncConfig.class) { + if (esSyncConfig == null) { + esSyncConfig = ESSyncConfigLoader.load(); + } + } + } + Map properties = configuration.getProperties(); + Settings.Builder settingBuilder = Settings.builder(); + properties.forEach(settingBuilder::put); + Settings settings = settingBuilder.build(); + transportClient = new PreBuiltTransportClient(settings); + esSyncService = new ESSyncService(transportClient); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + + @Override + public void sync(Dml dml) { + } + + @Override + public EtlResult etl(String task, List params) { + return null; + } + + @Override + public Map count(String task) { + return null; + } + + @Override + public void destroy() { + + } + + @Override + public String getDestination(String task) { + return null; + } +} 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 new file mode 100644 index 00000000..be0b712b --- /dev/null +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfig.java @@ -0,0 +1,137 @@ +package com.alibaba.otter.canal.client.adapter.es.config; + +import java.util.ArrayList; +import java.util.List; + +public class ESSyncConfig { + + private String dataSourceKey; // 数据源key + + private ESMapping esMapping; + + public void validate() { + if (esMapping._index == null) { + throw new NullPointerException("esMapping._index"); + } + if (esMapping._type == null) { + throw new NullPointerException("esMapping._type"); + } + if (esMapping._id == null && esMapping.pk == null) { + throw new NullPointerException("esMapping._id and esMapping.pk"); + } + if (esMapping.sql == null) { + throw new NullPointerException("esMapping.sql"); + } + } + + public String getDataSourceKey() { + return dataSourceKey; + } + + public void setDataSourceKey(String dataSourceKey) { + this.dataSourceKey = dataSourceKey; + } + + public ESMapping getEsMapping() { + return esMapping; + } + + public void setEsMapping(ESMapping esMapping) { + this.esMapping = esMapping; + } + + public static class ESMapping { + + private String _index; + private String _type; + private String _id; + private String pk; + private String parent; + private String sql; + private List skips = new ArrayList<>(); + private boolean alwaysSql = false; + private int commitBatch = 1000; + private String etlCondition; + + public String get_index() { + return _index; + } + + public void set_index(String _index) { + this._index = _index; + } + + public String get_type() { + return _type; + } + + public void set_type(String _type) { + this._type = _type; + } + + public String get_id() { + return _id; + } + + public void set_id(String _id) { + this._id = _id; + } + + public String getPk() { + return pk; + } + + public void setPk(String pk) { + this.pk = pk; + } + + public String getParent() { + return parent; + } + + public void setParent(String parent) { + this.parent = parent; + } + + public List getSkips() { + return skips; + } + + public void setSkips(List skips) { + this.skips = skips; + } + + public String getSql() { + return sql; + } + + public void setSql(String sql) { + this.sql = sql; + } + + public boolean isAlwaysSql() { + return alwaysSql; + } + + public void setAlwaysSql(boolean alwaysSql) { + this.alwaysSql = alwaysSql; + } + + public int getCommitBatch() { + return commitBatch; + } + + public void setCommitBatch(int commitBatch) { + this.commitBatch = commitBatch; + } + + public String getEtlCondition() { + return etlCondition; + } + + public void setEtlCondition(String etlCondition) { + this.etlCondition = etlCondition; + } + + } +} diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfigLoader.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfigLoader.java new file mode 100644 index 00000000..7e156409 --- /dev/null +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfigLoader.java @@ -0,0 +1,89 @@ +package com.alibaba.otter.canal.client.adapter.es.config; + +import java.io.File; +import java.io.FileInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import java.util.Collection; +import java.util.LinkedHashMap; +import java.util.Map; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.yaml.snakeyaml.Yaml; + +import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs; + +public class ESSyncConfigLoader { + + private static Logger logger = LoggerFactory.getLogger(ESSyncConfigLoader.class); + + private static final String BASE_PATH = "es"; + + public static Map load() { + logger.info("## Start loading mapping config ... "); + + Map result = new LinkedHashMap<>(); + + Collection configs = AdapterConfigs.get("es"); + for (String c : configs) { + if (c == null) { + continue; + } + c = c.trim(); + if (c.equals("") || c.startsWith("#")) { + continue; + } + + ESSyncConfig config; + String configContent = null; + + if (c.endsWith(".yml")) { + configContent = readConfigContent(BASE_PATH + "/" + c); + } + + config = new Yaml().loadAs(configContent, ESSyncConfig.class); + + try { + config.validate(); + } catch (Exception e) { + throw new RuntimeException("ERROR Config: " + c + " " + e.getMessage(), e); + } + result.put(c, config); + } + + logger.info("## Mapping config loaded"); + return result; + } + + public static String readConfigContent(String config) { + InputStream in = null; + try { + // 先取本地文件，再取类路径 + File configFile = new File("config/" + config); + if (configFile.exists()) { + in = new FileInputStream(configFile); + } else { + in = ESSyncConfigLoader.class.getClassLoader().getResourceAsStream(config); + } + if (in == null) { + throw new RuntimeException("Config file not found."); + } + + byte[] bytes = new byte[in.available()]; + in.read(bytes); + return new String(bytes, StandardCharsets.UTF_8); + } catch (IOException e) { + throw new RuntimeException("Read yml config error ", e); + } finally { + try { + if (in != null) { + in.close(); + } + } catch (IOException e) { + // ignore + } + } + } +} 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 new file mode 100644 index 00000000..88e29eb6 --- /dev/null +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SchemaItem.java @@ -0,0 +1,364 @@ +package com.alibaba.otter.canal.client.adapter.es.config; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +public class SchemaItem { + + private Map aliasTableItems = new LinkedHashMap<>(); + private Map selectFields = new LinkedHashMap<>(); + private String sql; + + private volatile Map> tableItemAliases; + private volatile Map> columnFields; + + public void init() { + this.getTableItemAliases(); + this.getColumnFields(); + aliasTableItems.values().forEach(tableItem -> { + tableItem.getRelationTableFields(); + tableItem.getRelationSelectFields(); + }); + } + + public Map getAliasTableItems() { + return aliasTableItems; + } + + public void setAliasTableItems(Map aliasTableItems) { + this.aliasTableItems = aliasTableItems; + } + + public String getSql() { + return sql; + } + + public void setSql(String sql) { + this.sql = sql; + } + + public Map getSelectFields() { + return selectFields; + } + + public void setSelectFields(Map selectFields) { + this.selectFields = selectFields; + } + + public Map> getTableItemAliases() { + if (tableItemAliases == null) { + synchronized (SchemaItem.class) { + if (tableItemAliases == null) { + tableItemAliases = new LinkedHashMap<>(); + aliasTableItems.forEach((alias, tableItem) -> { + List aliases = tableItemAliases + .computeIfAbsent(tableItem.getTableName().toLowerCase(), k -> new ArrayList<>()); + aliases.add(tableItem); + }); + } + } + } + return tableItemAliases; + } + + public Map> getColumnFields() { + if (columnFields == null) { + synchronized (SchemaItem.class) { + if (columnFields == null) { + columnFields = new LinkedHashMap<>(); + getSelectFields() + .forEach((fieldName, fieldItem) -> fieldItem.getColumnItems().forEach(columnItem -> { + TableItem tableItem = getAliasTableItems().get(columnItem.getOwner()); + if (!tableItem.isSubQuery()) { + List fieldItems = columnFields.computeIfAbsent( + columnItem.getOwner() + "." + columnItem.getColumnName(), + k -> new ArrayList<>()); + fieldItems.add(fieldItem); + } else { + tableItem.getSubQueryFields().forEach(subQueryField -> { + List fieldItems = columnFields.computeIfAbsent( + columnItem.getOwner() + "." + subQueryField.getColumn().getColumnName(), + k -> new ArrayList<>()); + fieldItems.add(fieldItem); + }); + } + })); + } + } + } + return columnFields; + + } + + public TableItem getMainTable() { + if (!aliasTableItems.isEmpty()) { + return aliasTableItems.values().iterator().next(); + } else { + return null; + } + } + + public static class TableItem { + + private SchemaItem schemaItem; + + private String schema; + private String tableName; + private String alias; + private String subQuerySql; + private List subQueryFields = new ArrayList<>(); + private List relationFields = new ArrayList<>(); + + private boolean main; + private boolean subQuery; + + private volatile List relationTableFields; // 当前表关联条件字段 + private volatile List relationSelectFieldItem; // 关联条件字段在select中的对应字段 + + public TableItem(SchemaItem schemaItem){ + this.schemaItem = schemaItem; + } + + public SchemaItem getSchemaItem() { + return schemaItem; + } + + public void setSchemaItem(SchemaItem schemaItem) { + this.schemaItem = schemaItem; + } + + public String getSchema() { + return schema; + } + + public void setSchema(String schema) { + this.schema = schema; + } + + public String getTableName() { + return tableName; + } + + public void setTableName(String tableName) { + this.tableName = tableName; + } + + public String getAlias() { + return alias; + } + + public void setAlias(String alias) { + this.alias = alias; + } + + public String getSubQuerySql() { + return subQuerySql; + } + + public void setSubQuerySql(String subQuerySql) { + this.subQuerySql = subQuerySql; + } + + public boolean isMain() { + return main; + } + + public void setMain(boolean main) { + this.main = main; + } + + public boolean isSubQuery() { + return subQuery; + } + + public void setSubQuery(boolean subQuery) { + this.subQuery = subQuery; + } + + public List getSubQueryFields() { + return subQueryFields; + } + + public void setSubQueryFields(List subQueryFields) { + this.subQueryFields = subQueryFields; + } + + public List getRelationFields() { + return relationFields; + } + + public void setRelationFields(List relationFields) { + this.relationFields = relationFields; + } + + public List getRelationTableFields() { + if (relationTableFields == null) { + synchronized (SchemaItem.class) { + if (relationTableFields == null) { + relationTableFields = new ArrayList<>(); + getRelationFields().forEach(relationFieldsPair -> { + FieldItem leftFieldItem = relationFieldsPair.getLeftFieldItem(); + FieldItem rightFieldItem = relationFieldsPair.getRightFieldItem(); + if (getAlias().equals(leftFieldItem.getOwner())) { + relationTableFields.add(leftFieldItem); + } else if (getAlias().equals(rightFieldItem.getOwner())) { + relationTableFields.add(rightFieldItem); + } + }); + } + } + } + return relationTableFields; + } + + public List getRelationSelectFields() { + if (relationSelectFieldItem == null) { + synchronized (SchemaItem.class) { + if (relationSelectFieldItem == null) { + relationSelectFieldItem = new ArrayList<>(); + getRelationFields().forEach(relationFieldsPair -> { + FieldItem leftFieldItem = relationFieldsPair.getLeftFieldItem(); + List selectFieldItem = getSchemaItem().getColumnFields() + .get(leftFieldItem.getOwner() + "." + leftFieldItem.getColumn().getColumnName()); + if (selectFieldItem != null && !selectFieldItem.isEmpty()) { + relationSelectFieldItem.addAll(selectFieldItem); + } else { + FieldItem rightFieldItem = relationFieldsPair.getRightFieldItem(); + selectFieldItem = getSchemaItem().getColumnFields() + .get(rightFieldItem.getOwner() + "." + rightFieldItem.getColumn().getColumnName()); + if (selectFieldItem != null && !selectFieldItem.isEmpty()) { + relationSelectFieldItem.addAll(selectFieldItem); + } else { + throw new UnsupportedOperationException( + "Relation condition column must in select columns."); + } + } + }); + } + } + } + return relationSelectFieldItem; + } + } + + public static class RelationFieldsPair { + + private FieldItem leftFieldItem; + private FieldItem rightFieldItem; + + public RelationFieldsPair(FieldItem leftFieldItem, FieldItem rightFieldItem){ + this.leftFieldItem = leftFieldItem; + this.rightFieldItem = rightFieldItem; + } + + public FieldItem getLeftFieldItem() { + return leftFieldItem; + } + + public void setLeftFieldItem(FieldItem leftFieldItem) { + this.leftFieldItem = leftFieldItem; + } + + public FieldItem getRightFieldItem() { + return rightFieldItem; + } + + public void setRightFieldItem(FieldItem rightFieldItem) { + this.rightFieldItem = rightFieldItem; + } + } + + public static class FieldItem { + + private String fieldName; + private List columnItems = new ArrayList<>(); + private List owners = new ArrayList<>(); + + private boolean method; + private boolean binaryOp; + + public String getFieldName() { + return fieldName; + } + + public void setFieldName(String fieldName) { + this.fieldName = fieldName; + } + + public List getColumnItems() { + return columnItems; + } + + public void setColumnItems(List columnItems) { + this.columnItems = columnItems; + } + + public boolean isMethod() { + return method; + } + + public void setMethod(boolean method) { + this.method = method; + } + + public boolean isBinaryOp() { + return binaryOp; + } + + public void setBinaryOp(boolean binaryOp) { + this.binaryOp = binaryOp; + } + + public List getOwners() { + return owners; + } + + public void setOwners(List owners) { + this.owners = owners; + } + + public void addColumn(ColumnItem columnItem) { + columnItems.add(columnItem); + } + + public ColumnItem getColumn() { + if (!columnItems.isEmpty()) { + return columnItems.get(0); + } else { + return null; + } + } + + public String getOwner() { + if (!owners.isEmpty()) { + return owners.get(0); + } else { + return null; + } + } + } + + public static class ColumnItem { + + private String owner; + private String columnName; + + public String getOwner() { + return owner; + } + + public void setOwner(String owner) { + this.owner = owner; + } + + public String getColumnName() { + return columnName; + } + + public void setColumnName(String columnName) { + this.columnName = columnName; + } + } +} diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SqlParser.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SqlParser.java new file mode 100644 index 00000000..78fb93ec --- /dev/null +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SqlParser.java @@ -0,0 +1,205 @@ +package com.alibaba.otter.canal.client.adapter.es.config; + +import static com.alibaba.fastsql.sql.ast.expr.SQLBinaryOperator.BooleanAnd; +import static com.alibaba.fastsql.sql.ast.expr.SQLBinaryOperator.Equality; + +import java.util.ArrayList; +import java.util.List; +import java.util.stream.Collectors; + +import com.alibaba.fastsql.sql.SQLUtils; +import com.alibaba.fastsql.sql.ast.SQLExpr; +import com.alibaba.fastsql.sql.ast.expr.SQLBinaryOpExpr; +import com.alibaba.fastsql.sql.ast.expr.SQLIdentifierExpr; +import com.alibaba.fastsql.sql.ast.expr.SQLMethodInvokeExpr; +import com.alibaba.fastsql.sql.ast.expr.SQLPropertyExpr; +import com.alibaba.fastsql.sql.ast.statement.*; +import com.alibaba.fastsql.sql.dialect.mysql.ast.statement.MySqlSelectQueryBlock; +import com.alibaba.fastsql.sql.dialect.mysql.parser.MySqlStatementParser; +import com.alibaba.fastsql.sql.parser.ParserException; +import com.alibaba.fastsql.sql.parser.SQLStatementParser; +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.es.config.SchemaItem.RelationFieldsPair; +import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem.TableItem; + +/** + * ES同步指定sql格式解析 + * + * @author rewerma 2018-10-26 下午03:45:49 + * @version 1.0.0 + */ +public class SqlParser { + + /** + * 解析sql + * + * @param sql + * @return + */ + public static SchemaItem parse(String sql) { + try { + SQLStatementParser parser = new MySqlStatementParser(sql); + SQLSelectStatement statement = (SQLSelectStatement) parser.parseStatement(); + MySqlSelectQueryBlock sqlSelectQueryBlock = (MySqlSelectQueryBlock) statement.getSelect().getQuery(); + + SchemaItem schemaItem = new SchemaItem(); + schemaItem.setSql(SQLUtils.toMySqlString(sqlSelectQueryBlock)); + SQLTableSource sqlTableSource = sqlSelectQueryBlock.getFrom(); + List tableItems = new ArrayList<>(); + SqlParser.visitSelectTable(schemaItem, sqlTableSource, tableItems, null); + tableItems.forEach(tableItem -> schemaItem.getAliasTableItems().put(tableItem.getAlias(), tableItem)); + + List fieldItems = collectSelectQueryFields(sqlSelectQueryBlock); + fieldItems.forEach(fieldItem -> schemaItem.getSelectFields().put(fieldItem.getFieldName(), fieldItem)); + + schemaItem.init(); + + if (schemaItem.getAliasTableItems().isEmpty() || schemaItem.getSelectFields().isEmpty()) { + throw new ParserException("Parse sql error"); + } + return schemaItem; + } catch (Exception e) { + throw new ParserException(); + } + } + + /** + * 归集字段 + * + * @param sqlSelectQueryBlock + * @return + */ + private static List collectSelectQueryFields(MySqlSelectQueryBlock sqlSelectQueryBlock) { + return sqlSelectQueryBlock.getSelectList().stream().map(selectItem -> { + FieldItem fieldItem = new FieldItem(); + fieldItem.setFieldName(selectItem.getAlias()); + visitColumn(selectItem.getExpr(), fieldItem); + return fieldItem; + }).collect(Collectors.toList()); + } + + /** + * 解析字段 + * + * @param expr + * @param fieldItem + */ + private static void visitColumn(SQLExpr expr, FieldItem fieldItem) { + if (expr instanceof SQLIdentifierExpr) { + // 无owner + SQLIdentifierExpr identifierExpr = (SQLIdentifierExpr) expr; + if (fieldItem.getFieldName() == null) { + fieldItem.setFieldName(identifierExpr.getName()); + } + ColumnItem columnItem = new ColumnItem(); + columnItem.setColumnName(identifierExpr.getName()); + fieldItem.getOwners().add(null); + fieldItem.addColumn(columnItem); + } else if (expr instanceof SQLPropertyExpr) { + // 有owner + SQLPropertyExpr sqlPropertyExpr = (SQLPropertyExpr) expr; + if (fieldItem.getFieldName() == null) { + fieldItem.setFieldName(sqlPropertyExpr.getName()); + } + fieldItem.getOwners().add(sqlPropertyExpr.getOwnernName()); + ColumnItem columnItem = new ColumnItem(); + columnItem.setColumnName(sqlPropertyExpr.getName()); + columnItem.setOwner(sqlPropertyExpr.getOwnernName()); + fieldItem.addColumn(columnItem); + } else if (expr instanceof SQLMethodInvokeExpr) { + SQLMethodInvokeExpr methodInvokeExpr = (SQLMethodInvokeExpr) expr; + fieldItem.setMethod(true); + for (SQLExpr sqlExpr : methodInvokeExpr.getArguments()) { + visitColumn(sqlExpr, fieldItem); + } + } else if (expr instanceof SQLBinaryOpExpr) { + SQLBinaryOpExpr sqlBinaryOpExpr = (SQLBinaryOpExpr) expr; + fieldItem.setBinaryOp(true); + visitColumn(sqlBinaryOpExpr.getLeft(), fieldItem); + visitColumn(sqlBinaryOpExpr.getRight(), fieldItem); + } + } + + /** + * 解析表 + */ + private static void visitSelectTable(SchemaItem schemaItem, SQLTableSource sqlTableSource, + List tableItems, TableItem tableItemTmp) { + if (sqlTableSource instanceof SQLExprTableSource) { + SQLExprTableSource sqlExprTableSource = (SQLExprTableSource) sqlTableSource; + TableItem tableItem; + if (tableItemTmp != null) { + tableItem = tableItemTmp; + } else { + tableItem = new TableItem(schemaItem); + } + tableItem.setSchema(sqlExprTableSource.getSchema()); + tableItem.setTableName(sqlExprTableSource.getTableName()); + if (tableItem.getAlias() == null) { + tableItem.setAlias(sqlExprTableSource.getAlias()); + } + if (tableItems.isEmpty()) { + // 第一张表为主表 + tableItem.setMain(true); + } + tableItems.add(tableItem); + } else if (sqlTableSource instanceof SQLJoinTableSource) { + SQLJoinTableSource sqlJoinTableSource = (SQLJoinTableSource) sqlTableSource; + SQLTableSource leftTableSource = sqlJoinTableSource.getLeft(); + visitSelectTable(schemaItem, leftTableSource, tableItems, null); + SQLTableSource rightTableSource = sqlJoinTableSource.getRight(); + TableItem rightTableItem = new TableItem(schemaItem); + // 解析on条件字段 + visitOnCondition(sqlJoinTableSource.getCondition(), rightTableItem); + visitSelectTable(schemaItem, rightTableSource, tableItems, rightTableItem); + + } else if (sqlTableSource instanceof SQLSubqueryTableSource) { + SQLSubqueryTableSource subQueryTableSource = (SQLSubqueryTableSource) sqlTableSource; + MySqlSelectQueryBlock sqlSelectQuery = (MySqlSelectQueryBlock) subQueryTableSource.getSelect().getQuery(); + TableItem tableItem; + if (tableItemTmp != null) { + tableItem = tableItemTmp; + } else { + tableItem = new TableItem(schemaItem); + } + tableItem.setAlias(subQueryTableSource.getAlias()); + tableItem.setSubQuerySql(SQLUtils.toMySqlString(sqlSelectQuery)); + tableItem.setSubQuery(true); + tableItem.setSubQueryFields(collectSelectQueryFields(sqlSelectQuery)); + visitSelectTable(schemaItem, sqlSelectQuery.getFrom(), tableItems, tableItem); + } + } + + /** + * 解析on条件 + * + * @param expr + * @param tableItem + */ + private static void visitOnCondition(SQLExpr expr, TableItem tableItem) { + if (!(expr instanceof SQLBinaryOpExpr)) { + throw new UnsupportedOperationException(); + } + SQLBinaryOpExpr sqlBinaryOpExpr = (SQLBinaryOpExpr) expr; + if (sqlBinaryOpExpr.getOperator() == BooleanAnd) { + visitOnCondition(sqlBinaryOpExpr.getLeft(), tableItem); + visitOnCondition(sqlBinaryOpExpr.getRight(), tableItem); + } else if (sqlBinaryOpExpr.getOperator() == Equality) { + FieldItem leftFieldItem = new FieldItem(); + visitColumn(sqlBinaryOpExpr.getLeft(), leftFieldItem); + if (leftFieldItem.getColumnItems().size() != 1 || leftFieldItem.isMethod() || leftFieldItem.isBinaryOp()) { + throw new UnsupportedOperationException("Unsupported for complex of on-condition"); + } + FieldItem rightFieldItem = new FieldItem(); + visitColumn(sqlBinaryOpExpr.getRight(), rightFieldItem); + if (rightFieldItem.getColumnItems().size() != 1 || rightFieldItem.isMethod() + || rightFieldItem.isBinaryOp()) { + throw new UnsupportedOperationException("Unsupported for complex of on-condition"); + } + tableItem.getRelationFields().add(new RelationFieldsPair(leftFieldItem, rightFieldItem)); + } else { + throw new UnsupportedOperationException("Unsupported for complex of on-condition"); + } + } +} 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 new file mode 100644 index 00000000..c3eb1e3e --- /dev/null +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java @@ -0,0 +1,15 @@ +package com.alibaba.otter.canal.client.adapter.es.service; + +import org.elasticsearch.client.transport.TransportClient; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class ESSyncService { + private Logger logger = LoggerFactory.getLogger(this.getClass()); + + private TransportClient transportClient; + + public ESSyncService(TransportClient transportClient) { + this.transportClient = transportClient; + } +} diff --git a/client-adapter/elasticsearch/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.OuterAdapter b/client-adapter/elasticsearch/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.OuterAdapter new file mode 100644 index 00000000..b30310ee --- /dev/null +++ b/client-adapter/elasticsearch/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.OuterAdapter @@ -0,0 +1 @@ +es=com.alibaba.otter.canal.client.adapter.es.ESAdapter \ No newline at end of file diff --git a/client-adapter/elasticsearch/src/main/resources/es/myetst_user.yml b/client-adapter/elasticsearch/src/main/resources/es/myetst_user.yml new file mode 100644 index 00000000..df6239b6 --- /dev/null +++ b/client-adapter/elasticsearch/src/main/resources/es/myetst_user.yml @@ -0,0 +1,12 @@ +dataSourceKey: defaultDS +esMapping: + _index: mytest_user + _type: _doc + _id: id +# pk: id + sql: "select a.id, concat(a.name,'_test') as name, a.role_id, b.name as role_name, c.labels from user.a + left join role b on a.role_id=b.id + left join (select user_id, group_concat(label,',') as labels from user_label + group by user_id) c on c.user_id=a.id" + commitBatch: 3000 + etlCondition: "where a.c_time>='{0}' or b.c_time>='{0}' or c.c_time>='{0}'" \ No newline at end of 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 new file mode 100644 index 00000000..b2586632 --- /dev/null +++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/ConfigLoadTest.java @@ -0,0 +1,31 @@ +package com.alibaba.otter.canal.client.adapter.es.test; + +import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig; +import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfigLoader; +import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import java.util.Map; + +public class ConfigLoadTest { + + @Before + public void before() { + AdapterConfigs.put("es", "myetst_user.yml"); + } + + @Test + public void testLoad() { + Map configMap = ESSyncConfigLoader.load(); + ESSyncConfig config = configMap.get("myetst_user.yml"); + Assert.assertNotNull(config); + Assert.assertEquals("defaultDS", config.getDataSourceKey()); + ESSyncConfig.ESMapping esMapping = config.getEsMapping(); + Assert.assertEquals("mytest_user", esMapping.get_index()); + Assert.assertEquals("_doc", esMapping.get_type()); + Assert.assertEquals("id", esMapping.get_id()); + Assert.assertNotNull(esMapping.getSql()); + } +} 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 new file mode 100644 index 00000000..20025ad5 --- /dev/null +++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/SqlParseTest.java @@ -0,0 +1,51 @@ +package com.alibaba.otter.canal.client.adapter.es.test; + +import java.util.List; + +import org.junit.Assert; +import org.junit.Test; + +import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem; +import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem.FieldItem; +import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem.TableItem; +import com.alibaba.otter.canal.client.adapter.es.config.SqlParser; + +public class SqlParseTest { + + @Test + public void parseTest() { + // String sql = "select a.id,d.user_id2, concat(name,'_', a.nick) as name, + // b.name as roleNme, d.name as typeName,concat(d.label,'_') as label," + // + " c.name as refName from user a left join role b on b.user_id=a.id " + // + "left join type d on d.user_id=a.id and d.user_id2=a.p_id " + // + "left join ( select ref_id,group_concat(name,',') as name from role group + // by ref_id ) c on c.ref_id=a.id " + // + "where a.id=1"; + String sql = "select a.id, concat(a.name,'_test') as name, a.role_id, b.name as role_name, c.labels from user a " + + "left join role b on a.role_id=b.id " + + "left join (select user_id, group_concat(label,',') as labels from user_label " + + "group by user_id) c on c.user_id=a.id"; + SchemaItem schemaItem = SqlParser.parse(sql); + + // 通过表名找 TableItem + List tableItems = schemaItem.getTableItemAliases().get("user_label".toLowerCase()); + tableItems.forEach(tableItem -> Assert.assertEquals("c", tableItem.getAlias())); + + TableItem tableItem = tableItems.get(0); + Assert.assertFalse(tableItem.isMain()); + Assert.assertTrue(tableItem.isSubQuery()); + // 通过字段名找 FieldItem + List fieldItems = schemaItem.getColumnFields().get(tableItem.getAlias() + ".label".toLowerCase()); + fieldItems.forEach( + fieldItem -> Assert.assertEquals("c.labels", fieldItem.getOwner() + "." + fieldItem.getFieldName())); + + // 获取当前表关联条件字段 + List relationTableFields = tableItem.getRelationTableFields(); + relationTableFields.forEach(fieldItem -> Assert.assertEquals("user_id", fieldItem.getColumn().getColumnName())); + + // 获取关联字段在select中的对应字段 + List relationSelectFieldItem = tableItem.getRelationSelectFields(); + relationSelectFieldItem.forEach(fieldItem -> Assert.assertEquals("c.labels", + fieldItem.getOwner() + "." + fieldItem.getColumn().getColumnName())); + } +} diff --git a/client-adapter/pom.xml b/client-adapter/pom.xml index e2be0d35..b7f66111 100644 --- a/client-adapter/pom.xml +++ b/client-adapter/pom.xml @@ -21,6 +21,7 @@ common logger hbase + elasticsearch launcher