From 42a4abe4c7ff6abb0d57adfbac4afee955a88bf7 Mon Sep 17 00:00:00 2001 From: mcy Date: Thu, 25 Oct 2018 14:24:33 +0800 Subject: [PATCH] =?UTF-8?q?HBase=20etl=E9=94=81=E5=AE=9A=E5=90=8C=E6=AD=A5?= =?UTF-8?q?=E7=AD=89=E5=BE=85?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../canal/client/adapter/OuterAdapter.java | 10 ++++ .../client/adapter/hbase/HbaseAdapter.java | 9 ++++ .../adapter/hbase/config/MappingConfig.java | 9 ++++ .../main/resources/hbase/mytest_person2.yml | 1 + .../adapter/launcher/rest/CommonRest.java | 53 +++++++++++++++---- 5 files changed, 73 insertions(+), 9 deletions(-) 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 bf6d5fce..286ba398 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 @@ -55,4 +55,14 @@ public interface OuterAdapter { default Map count(String task) { throw new UnsupportedOperationException("unsupported operation"); } + + /** + * 通过task获取对应的destination + * + * @param task 任务名, 对应配置名 + * @return destination + */ + default String getDestination(String task) { + return null; + } } 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 22731960..50eb271c 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 @@ -162,4 +162,13 @@ public class HbaseAdapter implements OuterAdapter { } } } + + @Override + public String getDestination(String task) { + MappingConfig config = hbaseMapping.get(task); + if (config != null && config.getHbaseOrm() != null) { + return config.getHbaseOrm().getDestination(); + } + return null; + } } diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfig.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfig.java index c3c30c1b..a5ea5707 100644 --- a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfig.java +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfig.java @@ -146,6 +146,7 @@ public class MappingConfig { public static class HbaseOrm { private Mode mode = Mode.STRING; + private String destination; private String database; private String table; private String hbaseTable; @@ -170,6 +171,14 @@ public class MappingConfig { this.mode = mode; } + public String getDestination() { + return destination; + } + + public void setDestination(String destination) { + this.destination = destination; + } + public String getDatabase() { return database; } diff --git a/client-adapter/hbase/src/main/resources/hbase/mytest_person2.yml b/client-adapter/hbase/src/main/resources/hbase/mytest_person2.yml index cbf14d48..88cd99e2 100644 --- a/client-adapter/hbase/src/main/resources/hbase/mytest_person2.yml +++ b/client-adapter/hbase/src/main/resources/hbase/mytest_person2.yml @@ -1,6 +1,7 @@ dataSourceKey: defaultDS hbaseOrm: mode: PHOENIX #NATIVE #STRING + destination: example database: mytest # 数据库名 table: person2 # 数据库表名 hbaseTable: MYTEST.PERSON2 # HBase表名 diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/rest/CommonRest.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/rest/CommonRest.java index 4e4aa7cb..61c2b47b 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/rest/CommonRest.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/rest/CommonRest.java @@ -32,24 +32,43 @@ public class CommonRest { } /** - * Demo: curl http://127.0.0.1:8081/etl/hbase/mytest_person2.yml -X POST -d - * "params=0,1,2" + * ETL curl http://127.0.0.1:8081/etl/hbase/mytest_person2.yml -X POST + * + * @param type 类型 hbase, es + * @param task 任务名对应配置文件名 mytest_person2.yml + * @param params etl where条件参数, 为空全部导入 + * @return */ @PostMapping("/etl/{type}/{task}") public EtlResult etl(@PathVariable String type, @PathVariable String task, @RequestParam(name = "params", required = false) String params) { OuterAdapter adapter = loader.getExtension(type); - List paramArr = null; - if (params != null) { - String[] parmaArray = params.trim().split(";"); - paramArr = Arrays.asList(parmaArray); + String destination = adapter.getDestination(task); + Boolean oriSwithcStatus = null; + if (destination != null) { + oriSwithcStatus = syncSwitch.status(destination); + syncSwitch.off(destination); + } + try { + List paramArr = null; + if (params != null) { + String[] parmaArray = params.trim().split(";"); + paramArr = Arrays.asList(parmaArray); + } + return adapter.etl(task, paramArr); + } finally { + if (destination != null && oriSwithcStatus != null && oriSwithcStatus) { + syncSwitch.on(destination); + } } - - return adapter.etl(task, paramArr); } /** - * Demo: curl http://127.0.0.1:8081/count/hbase/mytest_person2.yml + * 统计总数 curl http://127.0.0.1:8081/count/hbase/mytest_person2.yml + * + * @param type 类型 hbase, es + * @param task 任务名对应配置文件名 mytest_person2.yml + * @return */ @GetMapping("/count/{type}/{task}") public Map count(@PathVariable String type, @PathVariable String task) { @@ -57,6 +76,9 @@ public class CommonRest { return adapter.count(task); } + /** + * 返回所有实例 curl http://127.0.0.1:8081/destinations + */ @GetMapping("/destinations") public List> destinations() { List> result = new ArrayList<>(); @@ -77,6 +99,13 @@ public class CommonRest { return result; } + /** + * 实例同步开关 curl http://127.0.0.1:8081/syncSwitch/example/off -X PUT + * + * @param destination 实例名称 + * @param status 开关状态: off on + * @return + */ @PutMapping("/syncSwitch/{destination}/{status}") public Result etl(@PathVariable String destination, @PathVariable String status) { if (status.equals("on")) { @@ -95,6 +124,12 @@ public class CommonRest { } } + /** + * 获取实例开关状态 curl http://127.0.0.1:8081/syncSwitch/example + * + * @param destination 实例名称 + * @return + */ @GetMapping("/syncSwitch/{destination}") public Map etl(@PathVariable String destination) { Boolean status = syncSwitch.status(destination);