Canal 是阿里巴巴集團(tuán)提供的一個(gè)開源產(chǎn)品，能夠通過解析數(shù)據(jù)庫的增量日志，提供增量數(shù)據(jù)的訂閱和消費(fèi)功能。使用 Canal 模擬成 MySQL 的 Slave，實(shí)時(shí)接收 MySQL 的增量數(shù)據(jù) binlog，然后通過 RESTful API 將數(shù)據(jù)寫入到 Easysearch 中。

前提條件

1. 部署 Easysearch 集群。
2. 部署 MySQL 數(shù)據(jù)庫。
3. 部署 Gateway，Canal Adapter 不支持使用 HTTPS 協(xié)議連接，使用 Gateway 代理 Easysearch 。
4. 部署 Console，方便查看 Easysearch 數(shù)據(jù)。

對(duì)于自建 MySQL , 需要先開啟 Binlog 寫入功能，配置 binlog-format 為 ROW 模式，my.cnf 中配置如下：

[mysqld]
log-bin=mysql-bin?#?開啟?binlog
binlog-format=ROW?#?選擇?ROW?模式
server_id=1?#?配置?MySQL?replaction?需要定義，不要和?canal?的?slaveId?重復(fù)

創(chuàng)建 canal 用戶，授權(quán) canal 連接 MySQL 具有作為 MySQL slave 的權(quán)限。

CREATE?USER?canal?IDENTIFIED?BY?'canal';
GRANT?SELECT,?REPLICATION?SLAVE,?REPLICATION?CLIENT?ON?*.*?TO?'canal'@'%';
--?GRANT?ALL?PRIVILEGES?ON?*.*?TO?'canal'@'%'?;
FLUSH?PRIVILEGES;

操作步驟

在進(jìn)行數(shù)據(jù)同步時(shí)支持自定義索引 Mapping，但需保證 Mapping 中定義的字段（名稱+類型）與 MySQL 中一致。

1. 準(zhǔn)備 MySQL 數(shù)據(jù)源

create?database?canal；
use?canal;
CREATE?TABLE?`test`?(
????`id`?bigint(32)?NOT?NULL,
????`name`?text?NOT?NULL,
????`age`?smallint??NOT?NULL,
????PRIMARY?KEY?(`id`)
)?ENGINE=InnoDB
DEFAULT?CHARACTER?SET=utf8;

2. Easysearch 創(chuàng)建索引

PUT?test
{
????"settings"?:?{
??????"index"?:?{
????????"number_of_shards"?:?"1",
????????"number_of_replicas"?:?"1"
??????}
????},
????"mappings"?:?{
????????????"properties"?:?{
??????????????"id":?{
???????????????????"type":?"integer"
???????????????},
???????????????"name":?{
????????????????????"type"?:?"text"
????????????????},
????????????????"age"?:?{
????????????????????"type"?:?"integer"
????????????????}
????????????}
????}
}

3. 安裝并啟動(dòng) Canal-server

下載：https://github.com/alibaba/canal/releases/download/canal-1.1.7/canal.deployer-1.1.7.tar.gz ?
修改配置文件
vi conf/example/instance.properties

4. 安裝并啟動(dòng) Canal-adapter

下載：https://github.com/alibaba/canal/releases/download/canal-1.1.7/canal.adapter-1.1.7.tar.gz ?
修改配置文件：application.yml

server:
??port:?8081
spring:
??jackson:
????date-format:?yyyy-MM-dd?HH:mm:ss
????time-zone:?GMT+8
????default-property-inclusion:?non_null

canal.conf:
??flatMessage:?true
??syncBatchSize:?1000
??retries:?-1
??timeout:
??accessKey:
??secretKey:
??consumerProperties:
????canal.tcp.server.host:?127.0.0.1:11111
????canal.tcp.batch.size:?500

??srcDataSources:
????defaultDS:
??????url:?jdbc:mysql://127.0.0.1:3306/canal?useUnicode=true
??????username:?canal
??????password:?canal
??canalAdapters:
????groups:
????-?groupId:?g1
??????outerAdapters:
??????-?name:?logger
??????-?name:?es7
????????properties:
??????????security.auth:?admin:4ad8f8f792e81cd0a6de
??????????cluster.name:?easysearch

新增 canal-adapter/conf/es7/test.yml，配置索引和表的映射關(guān)系。

dataSourceKey:?defaultDS
destination:?example
groupId:?g1
esMapping:
??_index:?test???????????#?es?的索引名稱
??_id:?_id???????????????#?es?的_id,?如果不配置該項(xiàng)必須配置下面的pk項(xiàng)_id則會(huì)由es自動(dòng)分配
??#?sql映射
??sql:?"?select?a.id?as?_id,a.id,a.name,a.age?from?test?a?"
??etlCondition:?"where?a.id>={}"
??commitBatch:?3000??????#?提交批大小

啟動(dòng) canal-adapter ?
./bin/startup.sh

5. 驗(yàn)證增量數(shù)據(jù)同步

在 MySQL 數(shù)據(jù)庫中，對(duì) test 表插入兩條數(shù)據(jù)。 ?
inserttest(id,name,age) values(1,'canal_test1',11); ?
inserttest(id,name,age) values(2,'canal_test2',22);

6. 在 Console 中，執(zhí)行以下命令查詢數(shù)據(jù)

最后

Canal 同步的是增量數(shù)據(jù)，不會(huì)同步之前的存量數(shù)據(jù)。要同步存量數(shù)據(jù)可參考《使用 Logstash 同步 MySQL 到 Easysearch》

關(guān)于 Easysearch

INFINI Easysearch 是一個(gè)分布式的近實(shí)時(shí)搜索與分析引擎，核心引擎基于開源的 Apache Lucene。Easysearch 的目標(biāo)是提供一個(gè)輕量級(jí)的 Elasticsearch 可替代版本，并繼續(xù)完善和支持更多的企業(yè)級(jí)功能。 與 Elasticsearch 相比，Easysearch 更關(guān)注在搜索業(yè)務(wù)場景的優(yōu)化和繼續(xù)保持其產(chǎn)品的簡潔與易用性。

官網(wǎng)文檔：https://www.infinilabs.com/docs/latest/easysearch

下載地址：https://www.infinilabs.com/download