From 49caff57599fe550cc55c30a65ed8dfe6f7b8dfb Mon Sep 17 00:00:00 2001 From: mcy Date: Wed, 24 Oct 2018 19:12:16 +0800 Subject: [PATCH] =?UTF-8?q?HBase=E7=9A=84etl=E5=88=9D=E6=AD=A5=E5=AE=8C?= =?UTF-8?q?=E6=88=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../canal/client/adapter/OuterAdapter.java | 5 ++- .../client/adapter/support/EtlResult.java | 37 ++++++++++++++++ .../client/adapter/hbase/HbaseAdapter.java | 14 ++++--- .../hbase/service/HbaseEtlService.java | 36 ++++++++-------- .../canal/adapter/launcher/rest/EtlRest.java | 42 +++++++++++++++++++ .../src/main/resources/application.yml | 6 ++- 6 files changed, 113 insertions(+), 27 deletions(-) create mode 100644 client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/EtlResult.java create mode 100644 client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/rest/EtlRest.java diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/OuterAdapter.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/OuterAdapter.java index a508d1ef..3fef6c0e 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/OuterAdapter.java +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/OuterAdapter.java @@ -2,8 +2,9 @@ package com.alibaba.otter.canal.client.adapter; import java.util.List; -import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig; 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; /** @@ -40,7 +41,7 @@ public interface OuterAdapter { * @param task 任务名, 对应配置名 * @param params etl筛选条件 */ - default void etl(String task, List params) { + default EtlResult etl(String task, List params) { throw new UnsupportedOperationException(); } diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/EtlResult.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/EtlResult.java new file mode 100644 index 00000000..0004e51a --- /dev/null +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/EtlResult.java @@ -0,0 +1,37 @@ +package com.alibaba.otter.canal.client.adapter.support; + +import java.io.Serializable; + +public class EtlResult implements Serializable { + private static final long serialVersionUID = 4250522736289866505L; + + private boolean succeeded = false; + + private String resultMessage; + + private String errorMessage; + + public boolean getSucceeded() { + return succeeded; + } + + public void setSucceeded(boolean succeeded) { + this.succeeded = succeeded; + } + + public String getResultMessage() { + return resultMessage; + } + + public void setResultMessage(String resultMessage) { + this.resultMessage = resultMessage; + } + + public String getErrorMessage() { + return errorMessage; + } + + public void setErrorMessage(String errorMessage) { + this.errorMessage = errorMessage; + } +} \ No newline at end of file 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 da51ac88..88c7fe22 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 @@ -18,10 +18,7 @@ import com.alibaba.otter.canal.client.adapter.hbase.config.MappingConfigLoader; import com.alibaba.otter.canal.client.adapter.hbase.service.HbaseEtlService; import com.alibaba.otter.canal.client.adapter.hbase.service.HbaseSyncService; import com.alibaba.otter.canal.client.adapter.hbase.support.HbaseTemplate; -import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; -import com.alibaba.otter.canal.client.adapter.support.Dml; -import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig; -import com.alibaba.otter.canal.client.adapter.support.SPI; +import com.alibaba.otter.canal.client.adapter.support.*; /** * HBase外部适配器 @@ -80,11 +77,16 @@ public class HbaseAdapter implements OuterAdapter { } @Override - public void etl(String task, List params) { + public EtlResult etl(String task, List params) { MappingConfig config = hbaseMapping.get(task); DataSource dataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey()); if (dataSource != null) { - HbaseEtlService.importData(dataSource, hbaseTemplate, config, params); + return HbaseEtlService.importData(dataSource, hbaseTemplate, config, params); + } else { + EtlResult etlResult = new EtlResult(); + etlResult.setSucceeded(false); + etlResult.setErrorMessage("DataSource not found"); + return etlResult; } } diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseEtlService.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseEtlService.java index 2c919955..e2b86df1 100644 --- a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseEtlService.java +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseEtlService.java @@ -20,6 +20,7 @@ import org.slf4j.LoggerFactory; import com.alibaba.otter.canal.client.adapter.hbase.config.MappingConfig; import com.alibaba.otter.canal.client.adapter.hbase.support.*; +import com.alibaba.otter.canal.client.adapter.support.EtlResult; import com.alibaba.otter.canal.client.adapter.support.JdbcTypeUtil; import com.google.common.base.Joiner; @@ -75,19 +76,19 @@ public class HbaseEtlService { } } - public static void importData(DataSource ds, HbaseTemplate hbaseTemplate, MappingConfig config, - List params) { - // EtlResult etlResult = new EtlResult(); + public static EtlResult importData(DataSource ds, HbaseTemplate hbaseTemplate, MappingConfig config, + List params) { + EtlResult etlResult = new EtlResult(); AtomicLong successCount = new AtomicLong(); List errMsg = new ArrayList<>(); String hbaseTable = ""; try { - // if (config == null) { - // logger.error("Config is null!"); - // etlResult.setSucceeded(false); - // etlResult.setErrorMessage("Config is null!"); - // return etlResult; - // } + if (config == null) { + logger.error("Config is null!"); + etlResult.setSucceeded(false); + etlResult.setErrorMessage("Config is null!"); + return etlResult; + } MappingConfig.HbaseOrm hbaseOrm = config.getHbaseOrm(); hbaseTable = hbaseOrm.getHbaseTable(); @@ -158,7 +159,7 @@ public class HbaseEtlService { // 当大于1万条记录时开启多线程 if (cnt >= 10000) { - int threadCount = 3; // TODO 从配置读取默认为3 + int threadCount = 3; long perThreadCnt = cnt / threadCount; ExecutorService executor = Executors.newFixedThreadPool(threadCount); List> futures = new ArrayList<>(threadCount); @@ -191,19 +192,18 @@ public class HbaseEtlService { logger.info( hbaseOrm.getHbaseTable() + " etl completed in: " + (System.currentTimeMillis() - start) / 1000 + "s!"); - // etlResult.setResultMessage("导入HBase表 " + hbaseOrm.getHbaseTable() + " 数据：" + - // successCount.get() + " 条"); + etlResult.setResultMessage("导入HBase表 " + hbaseOrm.getHbaseTable() + " 数据：" + successCount.get() + " 条"); } catch (Exception e) { logger.error(e.getMessage(), e); errMsg.add(hbaseTable + " etl failed! ==>" + e.getMessage()); } - // if (errMsg.isEmpty()) { - // etlResult.setSucceeded(true); - // } else { - // etlResult.setErrorMessage(Joiner.on("\n").join(errMsg)); - // } - // return etlResult; + if (errMsg.isEmpty()) { + etlResult.setSucceeded(true); + } else { + etlResult.setErrorMessage(Joiner.on("\n").join(errMsg)); + } + return etlResult; } private static boolean executeSqlImport(DataSource ds, String sql, MappingConfig.HbaseOrm hbaseOrm, diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/rest/EtlRest.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/rest/EtlRest.java new file mode 100644 index 00000000..33b1d8f9 --- /dev/null +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/rest/EtlRest.java @@ -0,0 +1,42 @@ +package com.alibaba.otter.canal.adapter.launcher.rest; + +import java.util.Arrays; +import java.util.List; + +import javax.annotation.PostConstruct; + +import com.alibaba.otter.canal.client.adapter.support.EtlResult; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.PathVariable; +import org.springframework.web.bind.annotation.RequestParam; +import org.springframework.web.bind.annotation.RestController; + +import com.alibaba.otter.canal.client.adapter.OuterAdapter; +import com.alibaba.otter.canal.client.adapter.support.ExtensionLoader; + +@RestController +public class EtlRest { + + private ExtensionLoader loader; + + @PostConstruct + public void init() { + loader = ExtensionLoader.getExtensionLoader(OuterAdapter.class); + } + + /** + * Demo: http://127.0.0.1:8081/etl/hbase/mytest_person2.yml + */ + @GetMapping("/etl/{type}/{task}") + public EtlResult etl(@PathVariable String type, @PathVariable String task, + @RequestParam(required = false) String params) { + OuterAdapter adapter = loader.getExtension(type); + List paramArr = null; + if (params != null) { + String[] parmaArray = params.trim().split(";"); + paramArr = Arrays.asList(parmaArray); + } + + return adapter.etl(task, paramArr); + } +} diff --git a/client-adapter/launcher/src/main/resources/application.yml b/client-adapter/launcher/src/main/resources/application.yml index af787a9f..7a319012 100644 --- a/client-adapter/launcher/src/main/resources/application.yml +++ b/client-adapter/launcher/src/main/resources/application.yml @@ -1,9 +1,13 @@ server: port: 8081 - logging: level: com.alibaba.otter.canal.client.adapter.hbase: DEBUG +spring: + jackson: + date-format: yyyy-MM-dd HH:mm:ss + time-zone: GMT+8 + default-property-inclusion: non_null canal.conf: canalServerHost: 127.0.0.1:11111