fixed format

This commit is contained in:
agapple
2020-03-05 15:26:59 +08:00
parent b2b2a82554
commit afbf125231
13 changed files with 253 additions and 256 deletions

View File

@@ -1,5 +1,16 @@
package com.alibaba.otter.canal.client.adapter.kudu;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.client.adapter.OuterAdapter;
import com.alibaba.otter.canal.client.adapter.kudu.config.KuduMappingConfig;
import com.alibaba.otter.canal.client.adapter.kudu.config.KuduMappingConfigLoader;
@@ -11,12 +22,6 @@ 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;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
/**
* @author liuyadong
@@ -24,6 +29,7 @@ import java.util.concurrent.ConcurrentHashMap;
*/
@SPI("kudu")
public class KuduAdapter implements OuterAdapter {
private static Logger logger = LoggerFactory.getLogger(KuduAdapter.class);
private Map<String, KuduMappingConfig> kuduMapping = new ConcurrentHashMap<>(); // 文件名对应配置
@@ -47,7 +53,6 @@ public class KuduAdapter implements OuterAdapter {
return mappingConfigCache;
}
@Override
public void init(OuterAdapterConfig configuration, Properties envProperties) {
this.envProperties = envProperties;
@@ -72,12 +77,10 @@ public class KuduAdapter implements OuterAdapter {
if (envProperties != null && !"tcp".equalsIgnoreCase(envProperties.getProperty("canal.conf.mode"))) {
k = StringUtils.trimToEmpty(mappingConfig.getDestination()) + "-"
+ StringUtils.trimToEmpty(mappingConfig.getGroupId()) + "_"
+ mappingConfig.getKuduMapping().getDatabase() + "-"
+ mappingConfig.getKuduMapping().getTable();
+ mappingConfig.getKuduMapping().getDatabase() + "-" + mappingConfig.getKuduMapping().getTable();
} else {
k = StringUtils.trimToEmpty(mappingConfig.getDestination()) + "_"
+ mappingConfig.getKuduMapping().getDatabase() + "-"
+ mappingConfig.getKuduMapping().getTable();
+ mappingConfig.getKuduMapping().getDatabase() + "-" + mappingConfig.getKuduMapping().getTable();
}
Map<String, KuduMappingConfig> configMap = mappingConfigCache.computeIfAbsent(k,
k1 -> new ConcurrentHashMap<>());
@@ -108,9 +111,7 @@ public class KuduAdapter implements OuterAdapter {
String table = dml.getTable();
Map<String, KuduMappingConfig> configMap;
if (envProperties != null && !"tcp".equalsIgnoreCase(envProperties.getProperty("canal.conf.mode"))) {
configMap = mappingConfigCache.get(destination + "-"
+ groupId + "_"
+ database + "-" + table);
configMap = mappingConfigCache.get(destination + "-" + groupId + "_" + database + "-" + table);
} else {
configMap = mappingConfigCache.get(destination + "_" + database + "-" + table);
}
@@ -131,7 +132,8 @@ public class KuduAdapter implements OuterAdapter {
logger.error("groupID didn't mach,please check your gruopId ");
}
} else {
logger.error("{} config didn't get,please check your map key ", destination + "_" + database + "-" + table);
logger.error("{} config didn't get,please check your map key ", destination + "_" + database + "-"
+ table);
}
}
}

View File

@@ -1,16 +1,18 @@
package com.alibaba.otter.canal.client.adapter.kudu.config;
import com.alibaba.otter.canal.client.adapter.support.AdapterConfig;
import org.apache.commons.lang.StringUtils;
import java.util.LinkedHashMap;
import java.util.Map;
import org.apache.commons.lang.StringUtils;
import com.alibaba.otter.canal.client.adapter.support.AdapterConfig;
/**
* @author liuyadong
* @description kudu配置文件映射
*/
public class KuduMappingConfig implements AdapterConfig {
private String dataSourceKey; // 数据源key
private String destination; // canal实例或MQ的topic
@@ -77,7 +79,6 @@ public class KuduMappingConfig implements AdapterConfig {
this.destination = destination;
}
public void validate() {
if (kuduMapping.database == null || kuduMapping.database.isEmpty()) {
throw new NullPointerException("KuduMapping.database");

View File

@@ -1,19 +1,21 @@
package com.alibaba.otter.canal.client.adapter.kudu.config;
import com.alibaba.otter.canal.client.adapter.config.YmlConfigBinder;
import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Properties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.client.adapter.config.YmlConfigBinder;
import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader;
/**
* @author liuyadong
* @description kudu表信息加载
*/
public class KuduMappingConfigLoader {
private static Logger logger = LoggerFactory.getLogger(KuduMappingConfigLoader.class);
/**
@@ -28,8 +30,11 @@ public class KuduMappingConfigLoader {
Map<String, String> configContentMap = MappingConfigsLoader.loadConfigs("kudu");
configContentMap.forEach((fileName, content) -> {
KuduMappingConfig config = YmlConfigBinder
.bindYmlToObj(null, content, KuduMappingConfig.class, null, envProperties);
KuduMappingConfig config = YmlConfigBinder.bindYmlToObj(null,
content,
KuduMappingConfig.class,
null,
envProperties);
if (config == null) {
return;
}

View File

@@ -1,10 +1,10 @@
package com.alibaba.otter.canal.client.adapter.kudu.monitor;
import com.alibaba.otter.canal.client.adapter.config.YmlConfigBinder;
import com.alibaba.otter.canal.client.adapter.kudu.KuduAdapter;
import com.alibaba.otter.canal.client.adapter.kudu.config.KuduMappingConfig;
import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader;
import com.alibaba.otter.canal.client.adapter.support.Util;
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;
import org.apache.commons.io.monitor.FileAlterationMonitor;
@@ -13,16 +13,18 @@ import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import com.alibaba.otter.canal.client.adapter.config.YmlConfigBinder;
import com.alibaba.otter.canal.client.adapter.kudu.KuduAdapter;
import com.alibaba.otter.canal.client.adapter.kudu.config.KuduMappingConfig;
import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader;
import com.alibaba.otter.canal.client.adapter.support.Util;
/**
* @author liuyadong
* @description 配置文件监听
*/
public class KuduConfigMonitor {
private static final Logger logger = LoggerFactory.getLogger(KuduConfigMonitor.class);
private static final String adapterName = "kudu";
@@ -64,6 +66,7 @@ public class KuduConfigMonitor {
* 配置文件监听
*/
private class FileListener extends FileAlterationListenerAdaptor {
@Override
public void onFileCreate(File file) {
super.onFileCreate(file);
@@ -71,8 +74,11 @@ public class KuduConfigMonitor {
try {
// 加载新增的配置文件
String configContent = MappingConfigsLoader.loadConfig(adapterName + File.separator + file.getName());
KuduMappingConfig config = YmlConfigBinder
.bindYmlToObj(null, configContent, KuduMappingConfig.class, null, envProperties);
KuduMappingConfig config = YmlConfigBinder.bindYmlToObj(null,
configContent,
KuduMappingConfig.class,
null,
envProperties);
if (config == null) {
return;
}
@@ -92,14 +98,17 @@ public class KuduConfigMonitor {
try {
if (kuduAdapter.getKuduMapping().containsKey(file.getName())) {
// 加载配置文件
String configContent = MappingConfigsLoader
.loadConfig(adapterName + File.separator + file.getName());
String configContent = MappingConfigsLoader.loadConfig(adapterName + File.separator
+ file.getName());
if (configContent == null) {
onFileDelete(file);
return;
}
KuduMappingConfig config = YmlConfigBinder
.bindYmlToObj(null, configContent, KuduMappingConfig.class, null, envProperties);
KuduMappingConfig config = YmlConfigBinder.bindYmlToObj(null,
configContent,
KuduMappingConfig.class,
null,
envProperties);
if (config == null) {
return;
}
@@ -114,7 +123,6 @@ public class KuduConfigMonitor {
}
}
@Override
public void onFileDelete(File file) {
super.onFileDelete(file);

View File

@@ -1,5 +1,18 @@
package com.alibaba.otter.canal.client.adapter.kudu.service;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
import javax.sql.DataSource;
import org.apache.kudu.client.KuduException;
import com.alibaba.otter.canal.client.adapter.kudu.config.KuduMappingConfig;
import com.alibaba.otter.canal.client.adapter.kudu.support.KuduTemplate;
import com.alibaba.otter.canal.client.adapter.kudu.support.SyncUtil;
@@ -8,13 +21,6 @@ import com.alibaba.otter.canal.client.adapter.support.AdapterConfig;
import com.alibaba.otter.canal.client.adapter.support.EtlResult;
import com.alibaba.otter.canal.client.adapter.support.Util;
import com.google.common.base.Joiner;
import org.apache.kudu.client.KuduException;
import javax.sql.DataSource;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.util.*;
import java.util.concurrent.atomic.AtomicLong;
/**
* @author liuyadong
@@ -49,9 +55,9 @@ public class KuduEtlService extends AbstractEtlService {
return importData(sql, params);
}
@Override
protected boolean executeSqlImport(DataSource ds, String sql, List<Object> values, AdapterConfig.AdapterMapping mapping, AtomicLong impCount, List<String> errMsg) {
protected boolean executeSqlImport(DataSource ds, String sql, List<Object> values,
AdapterConfig.AdapterMapping mapping, AtomicLong impCount, List<String> errMsg) {
KuduMappingConfig.KuduMapping kuduMapping = (KuduMappingConfig.KuduMapping) mapping;
// 获取字段元数据
Map<String, String> columnsMap = new LinkedHashMap<>();// 需要同步的字段

View File

@@ -1,28 +1,34 @@
package com.alibaba.otter.canal.client.adapter.kudu.service;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.kudu.client.KuduException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.otter.canal.client.adapter.kudu.config.KuduMappingConfig;
import com.alibaba.otter.canal.client.adapter.kudu.support.KuduTemplate;
import com.alibaba.otter.canal.client.adapter.support.Dml;
import org.apache.kudu.client.KuduException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
/**
* @author liuyadong
* @description kudu实时同步
*/
public class KuduSyncService {
private static Logger logger = LoggerFactory.getLogger(KuduSyncService.class);
private KuduTemplate kuduTemplate;
// 源库表字段类型缓存: instance.schema.table -> <columnName, jdbcType>
// private Map<String, Map<String, Integer>> columnsTypeCache = new ConcurrentHashMap<>();
// private Map<String, Map<String, Integer>> columnsTypeCache = new
// ConcurrentHashMap<>();
public KuduSyncService(KuduTemplate kuduTemplate){
this.kuduTemplate = kuduTemplate;
@@ -157,7 +163,6 @@ public class KuduSyncService {
}
}
}
/**

View File

@@ -1,16 +1,30 @@
package com.alibaba.otter.canal.client.adapter.kudu.support;
import org.apache.kudu.ColumnSchema;
import org.apache.kudu.Schema;
import org.apache.kudu.Type;
import org.apache.kudu.client.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.kudu.ColumnSchema;
import org.apache.kudu.Schema;
import org.apache.kudu.Type;
import org.apache.kudu.client.Delete;
import org.apache.kudu.client.Insert;
import org.apache.kudu.client.KuduClient;
import org.apache.kudu.client.KuduException;
import org.apache.kudu.client.KuduScanner;
import org.apache.kudu.client.KuduSession;
import org.apache.kudu.client.KuduTable;
import org.apache.kudu.client.OperationResponse;
import org.apache.kudu.client.PartialRow;
import org.apache.kudu.client.SessionConfiguration;
import org.apache.kudu.client.Upsert;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* @author liuyadong
@@ -39,10 +53,10 @@ public class KuduTemplate {
if (kuduClient == null) {
// kudu master 以逗号分隔
List<String> masterList = Arrays.asList(masters.split(","));
kuduClient = new KuduClient.KuduClientBuilder(masterList)
.defaultOperationTimeoutMs(60000)
kuduClient = new KuduClient.KuduClientBuilder(masterList).defaultOperationTimeoutMs(60000)
.defaultSocketReadTimeoutMs(60000)
.defaultAdminOperationTimeoutMs(60000).build();
.defaultAdminOperationTimeoutMs(60000)
.build();
}
}
@@ -192,7 +206,6 @@ public class KuduTemplate {
}
}
}
/**
@@ -207,7 +220,8 @@ public class KuduTemplate {
KuduTable kuduTable = kuduClient.openTable(tableName);// 打开表
KuduSession session = kuduClient.newSession(); // 创建写session,kudu必须通过session写入
try {
session.setFlushMode(SessionConfiguration.FlushMode.MANUAL_FLUSH); // 采取Flush方式 手动刷新
session.setFlushMode(SessionConfiguration.FlushMode.MANUAL_FLUSH); // 采取Flush方式
// 手动刷新
session.setMutationBufferSpace(OPERATION_BATCH);
// 获取元数据结构
Map<String, Type> metaMap = new HashMap<>();

View File

@@ -1,12 +1,13 @@
package com.alibaba.otter.canal.client.adapter.kudu.support;
import com.alibaba.otter.canal.client.adapter.kudu.config.KuduMappingConfig;
import org.apache.commons.lang.StringUtils;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import org.apache.commons.lang.StringUtils;
import com.alibaba.otter.canal.client.adapter.kudu.config.KuduMappingConfig;
/**
* @author liuyadong
* @description 工具

View File

@@ -1,27 +1,27 @@
package com.alibaba.otter.canal.client.adapter.kudu.test;
import java.util.Arrays;
import java.util.List;
import org.apache.kudu.client.KuduClient;
import org.apache.kudu.client.KuduException;
import org.apache.kudu.client.KuduTable;
import org.junit.Test;
import java.util.Arrays;
import java.util.List;
public class KuduConnectionTest {
@Test
public void test01() {
List<String> masterList = Arrays.asList("10.6.36.102:7051,10.6.36.187:7051,10.6.36.229:7051".split(","));
KuduClient kuduClient = new KuduClient.KuduClientBuilder(masterList).defaultOperationTimeoutMs(60000)
.defaultSocketReadTimeoutMs(30000).defaultAdminOperationTimeoutMs(60000).build();
.defaultSocketReadTimeoutMs(30000)
.defaultAdminOperationTimeoutMs(60000)
.build();
try {
List<String> tablesList = kuduClient.getTablesList().getTablesList();
System.out.println(tablesList.toString());
KuduTable web_white_list = kuduClient.openTable("web_white_list");
} catch (KuduException e) {
e.printStackTrace();
}

View File

@@ -1,36 +1,20 @@
package com.alibaba.otter.canal.client.adapter.kudu.test;
import com.alibaba.otter.canal.client.adapter.kudu.config.KuduMappingConfig;
import com.alibaba.otter.canal.client.adapter.kudu.config.KuduMappingConfigLoader;
import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
import java.util.Map;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import java.util.Map;
import com.alibaba.otter.canal.client.adapter.kudu.config.KuduMappingConfig;
import com.alibaba.otter.canal.client.adapter.kudu.config.KuduMappingConfigLoader;
import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
/**
* ━━━━━━神兽出没━━━━━━
*    ┏┓   ┏┓
*   ┏┛┻━━━┛┻┓
*   ┃   ━   ┃
*   ┃ ┳┛ ┗┳ ┃
*   ┃   ┻   ┃
*   ┗━┓   ┏━┛
*     ┃   ┃ 神兽保佑
*     ┃   ┃ 代码无bug
*     ┃   ┗━━━┓
*     ┃       ┣┓
*     ┃       ┏┛
*     ┗┓┓┏━┳┓┏┛
*      ┃┫┫ ┃┫┫
*      ┗┻┛ ┗┻┛
* ━━━━━━感觉萌萌哒━━━━━━
* Created by Liuyadong on 2019-11-13
*
* @description
*/
public class TestConfig {
@Before
public void before() {
// 加载数据源连接池

View File

@@ -1,9 +1,9 @@
package com.alibaba.otter.canal.client.adapter.kudu.test;
import com.alibaba.druid.pool.DruidDataSource;
import java.sql.SQLException;
import com.alibaba.druid.pool.DruidDataSource;
public class TestConstant {
public final static String jdbcUrl = "jdbc:mysql://10.0.9.5:3306/canal_manager?useUnicode=true";

View File

@@ -1,35 +1,18 @@
package com.alibaba.otter.canal.client.adapter.kudu.test.sync;
import java.util.HashMap;
import java.util.Map;
import com.alibaba.otter.canal.client.adapter.kudu.KuduAdapter;
import com.alibaba.otter.canal.client.adapter.kudu.test.TestConstant;
import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig;
import java.util.HashMap;
import java.util.Map;
/**
* ━━━━━━神兽出没━━━━━━
*    ┏┓   ┏┓
*   ┏┛┻━━━┛┻┓
*   ┃   ━   ┃
*   ┃ ┳┛ ┗┳ ┃
*   ┃   ┻   ┃
*   ┗━┓   ┏━┛
*     ┃   ┃ 神兽保佑
*     ┃   ┃ 代码无bug
*     ┃   ┗━━━┓
*     ┃       ┣┓
*     ┃       ┏┛
*     ┗┓┓┏━┳┓┏┛
*      ┃┫┫ ┃┫┫
*      ┗┻┛ ┗┻┛
* ━━━━━━感觉萌萌哒━━━━━━
* Created by Liuyadong on 2019-11-13
*
* @description
*/
public class Common {
public static KuduAdapter init() {
DatasourceConfig.DATA_SOURCES.put("defaultDS", TestConstant.dataSource);

View File

@@ -1,31 +1,19 @@
package com.alibaba.otter.canal.client.adapter.kudu.test.sync;
import com.alibaba.otter.canal.client.adapter.kudu.KuduAdapter;
import com.alibaba.otter.canal.client.adapter.support.Dml;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import org.junit.Before;
import org.junit.Test;
import java.util.*;
import com.alibaba.otter.canal.client.adapter.kudu.KuduAdapter;
import com.alibaba.otter.canal.client.adapter.support.Dml;
/**
* ━━━━━━神兽出没━━━━━━
*    ┏┓   ┏┓
*   ┏┛┻━━━┛┻┓
*   ┃   ━   ┃
*   ┃ ┳┛ ┗┳ ┃
*   ┃   ┻   ┃
*   ┗━┓   ┏━┛
*     ┃   ┃ 神兽保佑
*     ┃   ┃ 代码无bug
*     ┃   ┗━━━┓
*     ┃       ┣┓
*     ┃       ┏┛
*     ┗┓┓┏━┳┓┏┛
*      ┃┫┫ ┃┫┫
*      ┗┻┛ ┗┻┛
* ━━━━━━感觉萌萌哒━━━━━━
* Created by Liuyadong on 2019-11-13
*
* @description
*/
public class TestSyncKudu {