From 848cbe11dd024c8226544dbf59fed137a25e39e1 Mon Sep 17 00:00:00 2001 From: mcy Date: Fri, 26 Oct 2018 10:33:16 +0800 Subject: [PATCH 1/2] =?UTF-8?q?=E4=BB=A3=E7=A0=81=E6=95=B4=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../loader/AbstractCanalAdapterWorker.java | 120 ++++++++++-------- .../loader/CanalAdapterKafkaWorker.java | 49 +------ .../launcher/loader/CanalAdapterLoader.java | 14 +- .../loader/CanalAdapterRocketMQWorker.java | 68 ++++------ .../launcher/loader/CanalAdapterWorker.java | 55 +------- 5 files changed, 106 insertions(+), 200 deletions(-) diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java index cd818a9d..0486045c 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java @@ -2,20 +2,19 @@ package com.alibaba.otter.canal.adapter.launcher.loader; import java.util.ArrayList; import java.util.List; -import java.util.concurrent.Callable; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Future; -import com.alibaba.otter.canal.adapter.launcher.common.SyncSwitch; -import com.alibaba.otter.canal.adapter.launcher.config.SpringContext; -import com.alibaba.otter.canal.client.adapter.support.Dml; -import com.alibaba.otter.canal.protocol.FlatMessage; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.alibaba.otter.canal.adapter.launcher.common.SyncSwitch; +import com.alibaba.otter.canal.adapter.launcher.config.SpringContext; import com.alibaba.otter.canal.client.adapter.OuterAdapter; +import com.alibaba.otter.canal.client.adapter.support.Dml; import com.alibaba.otter.canal.client.adapter.support.MessageUtil; +import com.alibaba.otter.canal.protocol.FlatMessage; import com.alibaba.otter.canal.protocol.Message; /** @@ -44,21 +43,21 @@ public abstract class AbstractCanalAdapterWorker { protected void writeOut(final Message message) { List> futures = new ArrayList<>(); // 组间适配器并行运行 - for (List outerAdapters : canalOuterAdapters) { + canalOuterAdapters.forEach(outerAdapters -> { final List adapters = outerAdapters; futures.add(groupInnerExecutorService.submit(() -> { try { // 组内适配器穿行运行，尽量不要配置组内适配器 - for (final OuterAdapter c : adapters) { + adapters.forEach(adapter -> { long begin = System.currentTimeMillis(); - MessageUtil.parse4Dml(canalDestination, message, c::sync); + MessageUtil.parse4Dml(canalDestination, message, adapter::sync); if (logger.isDebugEnabled()) { logger.debug("{} elapsed time: {}", - c.getClass().getName(), + adapter.getClass().getName(), (System.currentTimeMillis() - begin)); } - } + }); return true; } catch (Exception e) { return false; @@ -67,79 +66,94 @@ public abstract class AbstractCanalAdapterWorker { // 等待所有适配器写入完成 // 由于是组间并发操作，所以将阻塞直到耗时最久的工作组操作完成 - for (Future f : futures) { + futures.forEach(future -> { try { - if (!f.get()) { + if (!future.get()) { logger.error("Outer adapter write failed"); } } catch (InterruptedException | ExecutionException e) { // ignore } - } - } + }); + }); } protected void writeOut(final FlatMessage flatMessage) { List> futures = new ArrayList<>(); // 组间适配器并行运行 - for (List outerAdapters : canalOuterAdapters) { - final List adapters = outerAdapters; - futures.add(groupInnerExecutorService.submit(new Callable() { - - @Override - public Boolean call() { - try { - // 组内适配器穿行运行，尽量不要配置组内适配器 - for (OuterAdapter c : adapters) { - long begin = System.currentTimeMillis(); - Dml dml = MessageUtil.flatMessage2Dml(canalDestination, flatMessage); - c.sync(dml); - if (logger.isDebugEnabled()) { - logger.debug("{} elapsed time: {}", - c.getClass().getName(), - (System.currentTimeMillis() - begin)); - } + canalOuterAdapters.forEach(outerAdapters -> { + futures.add(groupInnerExecutorService.submit(() -> { + try { + // 组内适配器穿行运行，尽量不要配置组内适配器 + outerAdapters.forEach(adapter -> { + long begin = System.currentTimeMillis(); + Dml dml = MessageUtil.flatMessage2Dml(canalDestination, flatMessage); + adapter.sync(dml); + if (logger.isDebugEnabled()) { + logger.debug("{} elapsed time: {}", + adapter.getClass().getName(), + (System.currentTimeMillis() - begin)); } - return true; - } catch (Exception e) { - return false; - } + }); + return true; + } catch (Exception e) { + return false; } })); // 等待所有适配器写入完成 // 由于是组间并发操作，所以将阻塞直到耗时最久的工作组操作完成 - for (Future f : futures) { + futures.forEach(future -> { try { - if (!f.get()) { + if (!future.get()) { logger.error("Outer adapter write failed"); } } catch (InterruptedException | ExecutionException e) { // ignore } - } + }); + }); + } + + public void start() { + if (!running) { + thread = new Thread(this::process); + thread.setUncaughtExceptionHandler(handler); + thread.start(); + running = true; } } - protected void stopOutAdapters() { - if (thread != null) { - try { - thread.join(); - } catch (InterruptedException e) { - // ignore + protected abstract void process(); + + public void stop() { + try { + if (!running) { + return; } - } - groupInnerExecutorService.shutdown(); - logger.info("topic connectors' worker thread dead!"); - for (List outerAdapters : canalOuterAdapters) { - for (OuterAdapter adapter : outerAdapters) { - adapter.destroy(); + + closeConnection(); + running = false; + + syncSwitch.release(canalDestination); + + logger.info("destination {} is waiting for adapters' worker thread die!", canalDestination); + if (thread != null) { + try { + thread.join(); + } catch (InterruptedException e) { + // ignore + } } + groupInnerExecutorService.shutdown(); + logger.info("destination {} adapters worker thread dead!", canalDestination); + canalOuterAdapters.forEach(outerAdapters -> outerAdapters.forEach(OuterAdapter::destroy)); + logger.info("destination {} all adapters destroyed!", canalDestination); + } catch (Exception e) { + logger.error(e.getMessage(), e); } - logger.info("topic all connectors destroyed!"); } - public abstract void start(); + protected abstract void closeConnection(); - public abstract void stop(); } diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java index 4c48125a..665ad38e 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java @@ -37,59 +37,18 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker { this.flatMessage = flatMessage; connector = KafkaCanalConnectors.newKafkaConnector(bootstrapServers, topic, null, groupId, flatMessage); // connector.setSessionTimeout(1L, TimeUnit.MINUTES); - - // super.initSwitcher(topic); } @Override - public void start() { - if (!running) { - thread = new Thread(() -> process()); - thread.setUncaughtExceptionHandler(handler); - running = true; - thread.start(); - } + protected void closeConnection(){ + connector.stopRunning(); } @Override - public void stop() { - try { - if (!running) { - return; - } - - connector.stopRunning(); - running = false; - - // if (switcher != null && !switcher.state()) { - // switcher.set(true); - // } - - if (thread != null) { - try { - thread.join(); - } catch (InterruptedException e) { - // ignore - } - } - groupInnerExecutorService.shutdown(); - logger.info("topic {} connectors' worker thread dead!", this.topic); - for (List outerAdapters : canalOuterAdapters) { - for (OuterAdapter adapter : outerAdapters) { - adapter.destroy(); - } - } - logger.info("topic {} all connectors destroyed!", this.topic); - } catch (Exception e) { - logger.error(e.getMessage(), e); - } - } - - private void process() { + protected void process() { while (!running) ; ExecutorService executor = Executors.newSingleThreadExecutor(); - // final AtomicBoolean executing = new AtomicBoolean(true); while (running) { try { logger.info("=============> Start to connect topic: {} <=============", this.topic); @@ -99,7 +58,7 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker { logger.info("=============> Subscribe topic: {} succeed <=============", this.topic); while (running) { try { - // switcher.get(); //等待开关开启 + syncSwitch.get(canalDestination); List messages; if (!flatMessage) { diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java index 3bb5b4a4..0447036a 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java @@ -135,20 +135,18 @@ public class CanalAdapterLoader { public void destroy() { if (canalWorkers.size() > 0) { ExecutorService stopExecutorService = Executors.newFixedThreadPool(canalWorkers.size()); - for (CanalAdapterWorker v : canalWorkers.values()) { - final CanalAdapterWorker caw = v; - stopExecutorService.submit(caw::stop); + for (CanalAdapterWorker canalAdapterWorker : canalWorkers.values()) { + stopExecutorService.submit(canalAdapterWorker::stop); } stopExecutorService.shutdown(); } if (canalMQWorker.size() > 0) { - ExecutorService stopMQWokerService = Executors.newFixedThreadPool(canalMQWorker.size()); - for (AbstractCanalAdapterWorker tmp : canalMQWorker.values()) { - final AbstractCanalAdapterWorker worker = tmp; - stopMQWokerService.submit(worker::stop); + ExecutorService stopMQWorkerService = Executors.newFixedThreadPool(canalMQWorker.size()); + for (AbstractCanalAdapterWorker canalAdapterMQWorker : canalMQWorker.values()) { + stopMQWorkerService.submit(canalAdapterMQWorker::stop); } - stopMQWokerService.shutdown(); + stopMQWorkerService.shutdown(); } logger.info("All canal adapters destroyed"); } diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java index 4632d7f0..f4c2e704 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java @@ -36,38 +36,12 @@ public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker { } @Override - public void start() { - if (!running) { - thread = new Thread(new Runnable() { - - @Override - public void run() { - process(); - } - }); - thread.setUncaughtExceptionHandler(handler); - running = true; - thread.start(); - } + protected void closeConnection() { + connector.stopRunning(); } @Override - public void stop() { - try { - if (!running) { - return; - } - connector.stopRunning(); - running = false; - logger.info("Stop topic {} out adapters begin", this.topic); - stopOutAdapters(); - logger.info("Stop topic {} out adapters end", this.topic); - } catch (Exception e) { - logger.error(e.getMessage(), e); - } - } - - private void process() { + protected void process() { while (!running) ; ExecutorService executor = Executors.newSingleThreadExecutor(); @@ -84,25 +58,27 @@ public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker { final Message message = connector.getWithoutAck(1); if (message != null) { - executor.submit(new Runnable() { - - @Override - public void run() { - try { - if (logger.isDebugEnabled()) { - logger.debug("topic: {} batchId: {} batchSize: {} ", topic, message.getId(), message.getEntries().size()); - } - long begin = System.currentTimeMillis(); - writeOut(message); - long now = System.currentTimeMillis(); - if ((System.currentTimeMillis() - begin) > 5 * 60 * 1000) { - logger.error("topic: {} batchId {} elapsed time: {} ms", topic, message.getId(), now - begin); - } - } catch (Exception e) { - logger.error(e.getMessage(), e); + executor.submit(() -> { + try { + if (logger.isDebugEnabled()) { + logger.debug("topic: {} batchId: {} batchSize: {} ", + topic, + message.getId(), + message.getEntries().size()); } - connector.ack(message.getId()); + long begin = System.currentTimeMillis(); + writeOut(message); + long now = System.currentTimeMillis(); + if ((System.currentTimeMillis() - begin) > 5 * 60 * 1000) { + logger.error("topic: {} batchId {} elapsed time: {} ms", + topic, + message.getId(), + now - begin); + } + } catch (Exception e) { + logger.error(e.getMessage(), e); } + connector.ack(message.getId()); }); } else { logger.debug("Message is null"); diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java index c728a15a..33a70f89 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java @@ -57,49 +57,12 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker { } @Override - public void start() { - if (!running) { - thread = new Thread(this::process); - thread.setUncaughtExceptionHandler(handler); - thread.start(); - running = true; - } + protected void closeConnection() { + connector.stopRunning(); } @Override - public void stop() { - try { - if (!running) { - return; - } - - connector.stopRunning(); - running = false; - - syncSwitch.release(canalDestination); - - logger.info("destination {} is waiting for adapters' worker thread die!", canalDestination); - if (thread != null) { - try { - thread.join(); - } catch (InterruptedException e) { - // ignore - } - } - groupInnerExecutorService.shutdown(); - logger.info("destination {} adapters' worker thread dead!", canalDestination); - for (List outerAdapters : canalOuterAdapters) { - for (OuterAdapter adapter : outerAdapters) { - adapter.destroy(); - } - } - logger.info("destination {} all adapters destroyed!", canalDestination); - } catch (Exception e) { - logger.error(e.getMessage(), e); - } - } - - private void process() { + protected void process() { while (!running) ; // waiting until running == true while (running) { @@ -129,11 +92,7 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker { int size = message.getEntries().size(); if (batchId == -1 || size == 0) { - try { - Thread.sleep(1000); - } catch (InterruptedException e) { - // ignore - } + Thread.sleep(1000); } else { if (logger.isDebugEnabled()) { logger.debug("destination: {} batchId: {} batchSize: {} ", @@ -143,18 +102,18 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker { } long begin = System.currentTimeMillis(); writeOut(message); - long now = System.currentTimeMillis(); if (logger.isDebugEnabled()) { logger.debug("destination: {} batchId: {} elapsed time: {} ms", this.canalDestination, batchId, - now - begin); + System.currentTimeMillis() - begin); } } connector.ack(batchId); // 提交确认 } catch (Exception e) { connector.rollback(batchId); // 处理失败, 回滚数据 - throw e; + logger.error("sync error!", e); + Thread.sleep(500); } } From bbfd4b499662c91662acdb81cd2439c0844a6a0a Mon Sep 17 00:00:00 2001 From: mcy Date: Fri, 26 Oct 2018 13:23:22 +0800 Subject: [PATCH 2/2] canal-adapter readme.md --- client-adapter/README.md | 251 ++++++++++++++++++ .../main/resources/hbase/mytest_person2.yml | 2 +- 2 files changed, 252 insertions(+), 1 deletion(-) create mode 100644 client-adapter/README.md diff --git a/client-adapter/README.md b/client-adapter/README.md new file mode 100644 index 00000000..47a5dcc4 --- /dev/null +++ b/client-adapter/README.md @@ -0,0 +1,251 @@ +## 基本说明 +canal 1.1.1版本之后, 增加客户端数据落地的适配及启动功能, 目前支持功能: +* 客户端启动器 +* 同步管理REST接口 +* 日志适配器, 作为DEMO +* HBase的数据同步(表对表同步), ETL功能 +* (后续支持) ElasticSearch多表数据同步,ETL功能 + +## 环境版本 +* 操作系统：无要求 +* java版本: jdk1.8 以上 +* canal 版本: 请下载最新的安装包，本文以当前v1.1.1 的canal.deployer-1.1.1.tar.gz为例 +* MySQL版本 ：5.7.18 +* HBase版本: Apache HBase 1.1.2, 若和服务端版本不一致可自行替换客户端HBase依赖 + +## 一、适配器启动器 +client-adapter分为适配器和启动器两部分, 适配器为多个fat jar, 每个适配器会将自己所需的依赖打成一个包, 以SPI的方式让启动器动态加载 + + +启动器为 SpringBoot 项目, 支持canal-client启动的同时提供相关REST管理接口, 运行目录结构为: +``` +canal-adapter-launcher.jar +- lib + client-adapter.logger-1.1.1-jar-with-dependencies.jar + client-adapter.hbase-1.1.1-jar-with-dependencies.jar +- config + application.yml + - hbase + mytest_person2.yml +``` +以上目录结构最终会打包成 canal-adapter-launcher.tar.gz 压缩包 + +## 二、启动器 +### 2.1 启动器配置 application.yml +#### canal相关配置部分说明 +``` +canal.conf: + canalServerHost: 127.0.0.1:11111 # 对应单机模式下的canal server的ip:port + zookeeperHosts: slave1:2181 # 对应集群模式下的zk地址, 如果配置了canalServerHost, 则以canalServerHost为准 + bootstrapServers: slave1:6667 # kafka或rocketMQ地址, 与canalServerHost不能并存 + flatMessage: true # 扁平message开关, 是否以json字符串形式投递数据, 仅在kafka/rocketMQ模式下有效 + canalInstances: # canal实例组, 如果是tcp模式可配置此项 + - instance: example # 对应canal destination + adapterGroups: # 对应适配器分组, 分组间的适配器并行运行 + - outAdapters: # 适配器列表, 分组内的适配串行运行 + - name: logger # 适配器SPI名 + - name: hbase + properties: # HBase相关连接参数 + hbase.zookeeper.quorum: slave1 + hbase.zookeeper.property.clientPort: 2181 + zookeeper.znode.parent: /hbase + mqTopics: # MQ topic租, 如果是kafka或者rockeMQ模式可配置此项, 与canalInstances不能并存 + - mqMode: kafka # MQ的模式: kafak/rocketMQ + topic: example # MQ topic + groups: # group组 + - groupId: g2 # group id + outAdapters: # 适配器列表, 以下配置和canalInstances中一样 + - name: logger +``` +#### 适配器相关配置部分说明 +``` +adapter.conf: + datasourceConfigs: # 数据源配置列表, 数据源将在适配器中用于ETL、数据同步回查等使用 + defaultDS: # 数据源 dataSourceKey, 适配器中通过该值获取指定数据源 + url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true + username: root + password: 121212 + adapterConfigs: # 适配器内部配置列表 + - hbase/mytest_person2.yml # 类型/配置文件名, 这里示例就是对应HBase适配器hbase目录下的mytest_person2.yml文件 +``` +## 2.2 适配器管理REST接口 +#### 2.2.1 查询所有订阅同步的canal destination或MQ topic +``` +curl http://127.0.0.1:8081/destinations +``` +#### 2.2.2 数据同步开关 +``` +curl http://127.0.0.1:8081/syncSwitch/example/off -X PUT +``` +针对 example 这个canal destination/MQ topic 进行开关操作. off代表关闭, 该destination/topic下的同步将阻塞或者断开连接不再接收数据, on代表开启 + +注: 如果在配置文件中配置了 zookeeperHosts 项, 则会使用分布式锁来控制HA中的数据同步开关, 如果是单机模式则使用本地锁来控制开关 +#### 2.2.3 数据同步开关状态 +``` +curl http://127.0.0.1:8081/syncSwitch/example +``` +查看指定 canal destination/MQ topic 的数据同步开关状态 +#### 2.2.4 手动ETL +``` +curl http://127.0.0.1:8081/etl/hbase/mytest_person2.yml -X POST -d "params=2018-10-21 00:00:00" +``` +导入数据到指定类型的库 +#### 2.2.5 查看相关库总数据 +``` +curl http://127.0.0.1:8081/count/hbase/mytest_person2.yml +``` +### 2.3 启动canal-adapter示例 +#### 2.3.1 启动canal server (单机模式), 参考: [Canal QuickStart](https://github.com/alibaba/canal/wiki/QuickStart) +#### 2.3.2 修改config/application.yml为: +``` +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 + flatMessage: true + canalInstances: + - instance: example + adapterGroups: + - outAdapters: + - name: logger +``` +启动 canal-adapter-launcher.jar +``` +java -jar canal-adapter-launcher.jar +``` + +## 三、HBase适配器 +### 3.1 修改启动器配置: application.yml +``` +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 + flatMessage: true + canalInstances: + - instance: example + adapterGroups: + - outAdapters: + - name: hbase + properties: + hbase.zookeeper.quorum: slave1 + hbase.zookeeper.property.clientPort: 2181 + zookeeper.znode.parent: /hbase + +adapter.conf: + datasourceConfigs: + defaultDS: + url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true + username: root + password: 121212 + adapterConfigs: + - hbase/mytest_person.yml +``` +其中指定了一个HBase表映射文件: mytest_person.yml +### 3.2 适配器表映射文件 +修改 config/hbase/mytest_person.yml文件: +``` +dataSourceKey: defaultDS # 对应application.yml中的datasourceConfigs下的配置 +hbaseOrm: # mysql--HBase的单表映射配置 + mode: STRING # HBase中的存储类型, 默认统一存为String, 可选: #PHOENIX #NATIVE #STRING + # NATIVE: 以java类型为主, PHOENIX: 将类型转换为Phoenix对应的类型 + destination: example # 对应 canal destination/MQ topic 名称 + database: mytest # 数据库名/schema名 + table: person # 表名 + hbaseTable: MYTEST.PERSON # HBase表名 + family: CF # 默认统一Column Family名称 + uppercaseQualifier: true # 字段名转大写, 默认为true + commitBatch: 3000 # 批量提交的大小, ETL中用到 + #rowKey: id,type # 复合字段rowKey不能和columns中的rowKey并存 + # 复合rowKey会以 '|' 分隔 + columns: # 字段映射, 如果不配置将自动映射所有字段, + # 并取第一个字段为rowKey, HBase字段名以mysql字段名为主 + id: ROWKE + name: CF:NAME + email: EMAIL # 如果column family为默认CF, 则可以省略 + type: # 如果HBase字段和mysql字段名一致, 则可以省略 + c_time: + birthday: +``` +如果涉及到类型转换,可以如下形式: +``` +... + columns: + id: ROWKE$STRING + ... + type: TYPE$BYTE + ... +``` +类型转换涉及到Java类型和Phoenix类型两种, 分别定义如下: +``` +#Java 类型转换, 对应配置 mode: NATIVE +$DEFAULT +$STRING +$INTEGER +$LONG +$SHORT +$BOOLEAN +$FLOAT +$DOUBLE +$BIGDECIMAL +$DATE +$BYTE +$BYTES +``` +``` +#Phoenix 类型转换, 对应配置 mode: PHOENIX +$DEFAULT 对应PHOENIX里的VARCHAR +$UNSIGNED_INT 对应PHOENIX里的UNSIGNED_INT 4字节 +$UNSIGNED_LONG 对应PHOENIX里的UNSIGNED_LONG 8字节 +$UNSIGNED_TINYINT 对应PHOENIX里的UNSIGNED_TINYINT 1字节 +$UNSIGNED_SMALLINT 对应PHOENIX里的UNSIGNED_SMALLINT 2字节 +$UNSIGNED_FLOAT 对应PHOENIX里的UNSIGNED_FLOAT 4字节 +$UNSIGNED_DOUBLE 对应PHOENIX里的UNSIGNED_DOUBLE 8字节 +$INTEGER 对应PHOENIX里的INTEGER 4字节 +$BIGINT 对应PHOENIX里的BIGINT 8字节 +$TINYINT 对应PHOENIX里的TINYINT 1字节 +$SMALLINT 对应PHOENIX里的SMALLINT 2字节 +$FLOAT 对应PHOENIX里的FLOAT 4字节 +$DOUBLE 对应PHOENIX里的DOUBLE 8字节 +$BOOLEAN 对应PHOENIX里的BOOLEAN 1字节 +$TIME 对应PHOENIX里的TIME 8字节 +$DATE 对应PHOENIX里的DATE 8字节 +$TIMESTAMP 对应PHOENIX里的TIMESTAMP 12字节 +$UNSIGNED_TIME 对应PHOENIX里的UNSIGNED_TIME 8字节 +$UNSIGNED_DATE 对应PHOENIX里的UNSIGNED_DATE 8字节 +$UNSIGNED_TIMESTAMP 对应PHOENIX里的UNSIGNED_TIMESTAMP 12字节 +$VARCHAR 对应PHOENIX里的VARCHAR 动态长度 +$VARBINARY 对应PHOENIX里的VARBINARY 动态长度 +$DECIMAL 对应PHOENIX里的DECIMAL 动态长度 +``` +如果不配置将以java对象原生类型默认映射转换 +### 3.3 启动HBase数据同步 +#### 创建HBase表 +在HBase shell中运行: +``` +create 'MYTEST.PERSON', {NAME=>'CF'} +``` +#### 启动canal-adapter启动器 +``` +java -jar canal-adapter-launcher.jar +``` +#### 验证 +修改mysql mytest.person表的数据, 将会自动同步到HBase的MYTEST.PERSON表下面, 并会打出DML的log \ No newline at end of file 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 88cd99e2..b22ffe5c 100644 --- a/client-adapter/hbase/src/main/resources/hbase/mytest_person2.yml +++ b/client-adapter/hbase/src/main/resources/hbase/mytest_person2.yml @@ -45,7 +45,7 @@ hbaseOrm: # $TINYINT 对应PHOENIX里的TINYINT 1字节 # $SMALLINT 对应PHOENIX里的SMALLINT 2字节 # $FLOAT 对应PHOENIX里的FLOAT 4字节 -# DOUBLE 对应PHOENIX里的DOUBLE 8字节 +# $DOUBLE 对应PHOENIX里的DOUBLE 8字节 # $BOOLEAN 对应PHOENIX里的BOOLEAN 1字节 # $TIME 对应PHOENIX里的TIME 8字节 # $DATE 对应PHOENIX里的DATE 8字节