chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

rdb etl count destination 接口适配

Browse Source 
es hbase 多源适配
This commit is contained in:
mcy
2018-11-12 15:07:46 +08:00
parent 7cca8857ff
commit 66912d3d05
18 changed files with 333 additions and 208 deletions
Download Patch File Download Diff File Expand all files Collapse all files

66
client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java
View File

@@ -1,9 +1,12 @@
package com.alibaba.otter.canal.client.adapter.es;
import java.net.InetAddress;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import javax.sql.DataSource;
@@ -13,10 +16,13 @@ import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.TransportAddress;
import org.elasticsearch.transport.client.PreBuiltTransportClient;
import com.alibaba.druid.pool.DruidDataSource;
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.ESSyncConfig.ESMapping;
import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfigLoader;
import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem;
import com.alibaba.otter.canal.client.adapter.es.config.SqlParser;
import com.alibaba.otter.canal.client.adapter.es.service.ESEtlService;
import com.alibaba.otter.canal.client.adapter.es.service.ESSyncService;
import com.alibaba.otter.canal.client.adapter.es.support.ESTemplate;
@@ -31,9 +37,12 @@ import com.alibaba.otter.canal.client.adapter.support.*;
@SPI("es")
public class ESAdapter implements OuterAdapter {
private TransportClient transportClient;
private Map<String, ESSyncConfig> esSyncConfig = new LinkedHashMap<>(); // 文件名对应配置
private Map<String, List<ESSyncConfig>> dbTableEsSyncConfig = new LinkedHashMap<>(); // schema-table对应配置
private ESSyncService esSyncService;
private TransportClient transportClient;
private ESSyncService esSyncService;
public TransportClient getTransportClient() {
return transportClient;
@@ -43,10 +52,48 @@ public class ESAdapter implements OuterAdapter {
return esSyncService;
}
public Map<String, ESSyncConfig> getEsSyncConfig() {
return esSyncConfig;
}
public Map<String, List<ESSyncConfig>> getDbTableEsSyncConfig() {
return dbTableEsSyncConfig;
}
@Override
public void init(OuterAdapterConfig configuration) {
try {
ESSyncConfigLoader.load();
SPI spi = this.getClass().getAnnotation(SPI.class);
Map<String, ESSyncConfig> esSyncConfigTmp = ESSyncConfigLoader.load(spi.value());
// 过滤不匹配的key的配置
esSyncConfigTmp.forEach((key, config) -> {
if ((config.getOuterAdapterKey() == null && configuration.getKey() == null)
|| config.getOuterAdapterKey().equalsIgnoreCase(configuration.getKey())) {
esSyncConfig.put(key, config);
}
});
for (ESSyncConfig config : esSyncConfig.values()) {
SchemaItem schemaItem = SqlParser.parse(config.getEsMapping().getSql());
config.getEsMapping().setSchemaItem(schemaItem);
DruidDataSource dataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
if (dataSource == null || dataSource.getUrl() == null) {
throw new RuntimeException("No data source found: " + config.getDataSourceKey());
}
Pattern pattern = Pattern.compile(".*:(.*)://.*/(.*)\\?.*$");
Matcher matcher = pattern.matcher(dataSource.getUrl());
if (!matcher.find()) {
throw new RuntimeException("Not found the schema of jdbc-url: " + config.getDataSourceKey());
}
String schema = matcher.group(2);
schemaItem.getAliasTableItems().values().forEach(tableItem -> {
List<ESSyncConfig> esSyncConfigs = dbTableEsSyncConfig
.computeIfAbsent(schema + "-" + tableItem.getTableName(), k -> new ArrayList<>());
esSyncConfigs.add(config);
});
}
Map<String, String> properties = configuration.getProperties();
Settings.Builder settingBuilder = Settings.builder();
@@ -68,13 +115,16 @@ public class ESAdapter implements OuterAdapter {
@Override
public void sync(Dml dml) {
esSyncService.sync(dml);
String database = dml.getDatabase();
String table = dml.getTable();
List<ESSyncConfig> esSyncConfigs = dbTableEsSyncConfig.get(database + "-" + table);
esSyncService.sync(esSyncConfigs, dml);
}
@Override
public EtlResult etl(String task, List<String> params) {
EtlResult etlResult = new EtlResult();
ESSyncConfig config = ESSyncConfigLoader.getEsSyncConfig().get(task);
ESSyncConfig config = esSyncConfig.get(task);
if (config != null) {
DataSource dataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
ESEtlService esEtlService = new ESEtlService(transportClient, config);
@@ -89,7 +139,7 @@ public class ESAdapter implements OuterAdapter {
StringBuilder resultMsg = new StringBuilder();
boolean resSuccess = true;
// ds不为空说明传入的是datasourceKey
for (ESSyncConfig configTmp : ESSyncConfigLoader.getEsSyncConfig().values()) {
for (ESSyncConfig configTmp : esSyncConfig.values()) {
// 取所有的destination为task的配置
if (configTmp.getDestination().equals(task)) {
ESEtlService esEtlService = new ESEtlService(transportClient, configTmp);
@@ -119,7 +169,7 @@ public class ESAdapter implements OuterAdapter {
@Override
public Map<String, Object> count(String task) {
ESSyncConfig config = ESSyncConfigLoader.getEsSyncConfig().get(task);
ESSyncConfig config = esSyncConfig.get(task);
ESMapping mapping = config.getEsMapping();
SearchResponse response = transportClient.prepareSearch(mapping.get_index())
.setTypes(mapping.get_type())
@@ -142,7 +192,7 @@ public class ESAdapter implements OuterAdapter {
@Override
public String getDestination(String task) {
ESSyncConfig config = ESSyncConfigLoader.getEsSyncConfig().get(task);
ESSyncConfig config = esSyncConfig.get(task);
if (config != null) {
return config.getDestination();
}

24
client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfig.java
View File

@@ -13,9 +13,11 @@ import java.util.Map;
*/
public class ESSyncConfig {
private String dataSourceKey; // 数据源key
private String dataSourceKey; // 数据源key
private String destination; // canal destination
private String outerAdapterKey; // adapter key
private String destination; // canal destination
private ESMapping esMapping;
@@ -42,6 +44,14 @@ public class ESSyncConfig {
this.dataSourceKey = dataSourceKey;
}
public String getOuterAdapterKey() {
return outerAdapterKey;
}
public void setOuterAdapterKey(String outerAdapterKey) {
this.outerAdapterKey = outerAdapterKey;
}
public String getDestination() {
return destination;
}
@@ -67,15 +77,15 @@ public class ESSyncConfig {
private String parent;
private String sql;
// 对象字段, 例: objFields:
// - _labels: array:;
private Map<String, String> objFields = new LinkedHashMap<>();
// - _labels: array:;
private Map<String, String> objFields = new LinkedHashMap<>();
private List<String> skips = new ArrayList<>();
private int commitBatch = 1000;
private String etlCondition;
private boolean syncByTimestamp = false; // 是否按时间戳定时同步
private Long syncInterval; // 同步时间间隔
private boolean syncByTimestamp = false; // 是否按时间戳定时同步
private Long syncInterval; // 同步时间间隔
private SchemaItem schemaItem; // sql解析结果模型
private SchemaItem schemaItem; // sql解析结果模型
public String get_index() {
return _index;

42
client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfigLoader.java
View File

@@ -32,24 +32,14 @@ public class ESSyncConfigLoader {
private static Logger logger = LoggerFactory
.getLogger(ESSyncConfigLoader.class);
private static final String BASE_PATH = "es";
private static volatile Map<String, ESSyncConfig> esSyncConfig = new LinkedHashMap<>(); // 文件名对应配置
private static volatile Map<String, List<ESSyncConfig>> dbTableEsSyncConfig = new LinkedHashMap<>(); // schema-table对应配置
public static Map<String, ESSyncConfig> getEsSyncConfig() {
return esSyncConfig;
}
public static Map<String, List<ESSyncConfig>> getDbTableEsSyncConfig() {
return dbTableEsSyncConfig;
}
public static synchronized void load() {
public static synchronized Map<String, ESSyncConfig> load(String name) {
logger.info("## Start loading es mapping config ... ");
Map<String, ESSyncConfig> esSyncConfig = new LinkedHashMap<>();
Collection<String> configs = AdapterConfigs.get("es");
if (configs == null) {
return;
return esSyncConfig;
}
for (String c : configs) {
if (c == null) {
@@ -64,32 +54,13 @@ public class ESSyncConfigLoader {
String configContent = null;
if (c.endsWith(".yml")) {
configContent = readConfigContent(BASE_PATH + "/" + c);
configContent = readConfigContent(name + "/" + c);
}
config = new Yaml().loadAs(configContent, ESSyncConfig.class);
try {
config.validate();
SchemaItem schemaItem = SqlParser.parse(config.getEsMapping().getSql());
config.getEsMapping().setSchemaItem(schemaItem);
DruidDataSource dataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
if (dataSource == null || dataSource.getUrl() == null) {
throw new RuntimeException("No data source found: " + config.getDataSourceKey());
}
Pattern pattern = Pattern.compile(".*:(.*)://.*/(.*)\\?.*$");
Matcher matcher = pattern.matcher(dataSource.getUrl());
if (!matcher.find()) {
throw new RuntimeException("Not found the schema of jdbc-url: " + config.getDataSourceKey());
}
String schema = matcher.group(2);
schemaItem.getAliasTableItems().values().forEach(tableItem -> {
List<ESSyncConfig> esSyncConfigs = dbTableEsSyncConfig
.computeIfAbsent(schema + "-" + tableItem.getTableName(), k -> new ArrayList<>());
esSyncConfigs.add(config);
});
} catch (Exception e) {
throw new RuntimeException("ERROR Config: " + c, e);
}
@@ -97,6 +68,7 @@ public class ESSyncConfigLoader {
}
logger.info("## ES mapping config loaded");
return esSyncConfig;
}
private static String readConfigContent(String config) {

5
client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java
View File

@@ -39,11 +39,8 @@ public class ESSyncService {
this.esTemplate = esTemplate;
}
public void sync(Dml dml) {
public void sync(List<ESSyncConfig> esSyncConfigs, Dml dml) {
long begin = System.currentTimeMillis();
String database = dml.getDatabase();
String table = dml.getTable();
List<ESSyncConfig> esSyncConfigs = ESSyncConfigLoader.getDbTableEsSyncConfig().get(database + "-" + table);
if (esSyncConfigs != null) {
if (logger.isTraceEnabled()) {
logger.trace("Destination: {}, database:{}, table:{}, type:{}, effect index count: {}",