Canal は、阿里巴巴（アリババ）グループが提供するオープンソース製品で、データベースの増分ログを解析することで、増分データのサブスクリプションと消費機能を提供します。Canal を MySQL のスレーブとして機能させ、MySQL の増分データ binlog をリアルタイムで受信し、その後 RESTful API を介して Easysearch にデータを書き込みます。
前提条件
- Easysearch クラスタのデプロイ。
- MySQL データベースのデプロイ。
- Gateway のデプロイ (Canal Adapter は HTTPS プロトコルをサポートしていないため、Easysearch への接続に Gateway をプロキシとして使用します)。
- Console のデプロイ (Easysearch データの確認に便利です)。
自建の MySQL では、まず binlog の書き込み機能を有効にする必要があります。binlog-format を ROW モードに設定します。my.cnf での設定例は以下の通りです：
[mysqld]
log-bin=mysql-bin # binlog を有効にする
binlog-format=ROW # ROW モードを選択
server_id=1 # MySQL レプリケーションを定義する必要があり、canal の slaveId と重複しないようにしてください
Canal ユーザーを作成し、MySQL スレーブとして接続するための権限を付与します。
CREATE USER canal_user IDENTIFIED BY 'canal_password';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal_user'@'%';
-- GRANT ALL PRIVILEGES ON *.* TO 'canal_user'@'%';
FLUSH PRIVILEGES;
操作手順
データ同期を行う際、カスタムインデックス Mapping をサポートしていますが、Mapping で定義されたフィールド（名前+タイプ）が MySQL と一致している必要があります。
1. MySQL データソースの準備
create database product_db;
use product_db;
CREATE TABLE `product` (
`product_id` bigint(32) NOT NULL,
`product_name` text NOT NULL,
`price` decimal(10, 2) NOT NULL,
PRIMARY KEY (`product_id`)
) ENGINE=InnoDB
DEFAULT CHARACTER SET=utf8;
2. Easysearch にインデックスを作成
PUT products
{
"settings" : {
"index" : {
"number_of_shards" : "1",
"number_of_replicas" : "1"
}
},
"mappings" : {
"properties" : {
"product_id": {
"type": "integer"
},
"product_name": {
"type" : "text"
},
"price" : {
"type" : "float"
}
}
}
}
3. 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
canal の起動
sh bin/startup.sh
起動成功のログ情報は、logs/canal/canal.log にあります。
canal の停止
sh bin/stop.sh
4. Canal-adapter のインストールと起動
ダウンロード：https://github.com/alibaba/canal/releases/download/canal-1.1.7/canal.adapter-1.1.7.tar.gz
設定ファイルの変更：application.yml
server:
port: 8082
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:11112
canal.tcp.batch.size: 500
srcDataSources:
defaultSource:
url: jdbc:mysql://127.0.0.1:3306/product_db?useUnicode=true
username: canal_user
password: canal_password
canalAdapters:
groups:
- groupId: sync_group
outerAdapters:
- name: logger
- name: es7
properties:
security.auth: admin:4ad8f8f792e81cd0a6de
cluster.name: easysearch
新規作成 canal-adapter/conf/es7/product.yml，インデックスとテーブルのマッピング関係を設定します。
dataSourceKey: defaultSource
destination: example_instance
groupId: sync_group
esMapping:
_index: products # es のインデックス名
_id: product_id # es の_id, この項目が設定されていない場合、pk 項目を設定する必要があります。_id は es によって自動的に割り当てられます。
# sql マッピング
sql: " select a.product_id as _id, a.product_id, a.product_name, a.price from product a "
etlCondition: "where a.product_id>={}"
commitBatch: 3000 # コミットバッチサイズ
canal-adapter の起動
./bin/startup.sh
5. 増分データ同期の検証
MySQL データベースで、`product` テーブルにデータを挿入します。
insert into `product`(`product_id`,`product_name`,`price`) values(101,'Smartphone',599.99);
insert into `product`(`product_id`,`product_name`,`price`) values(102,'Laptop',1299.99);
6. Console でデータをクエリする
Console で以下のコマンドを実行してデータをクエリします。
Canal は増分データのみを同期し、以前の既存データは同期しません。既存データを同期するには、「Logstash を使用して MySQL を Easysearch に同期する」を参照してください。
Easysearch について
INFINI Easysearch は、分散型のほぼリアルタイム検索および分析エンジンであり、コアエンジンはオープンソースの Apache Lucene に基づいています。Easysearch の目標は、軽量の Elasticsearch の代替バージョンを提供し、より多くのエンタープライズ機能を継続的に改善およびサポートすることです。Elasticsearch と比較して、Easysearch は検索ビジネスシナリオの最適化により焦点を当て、製品のシンプルさと使いやすさを維持することに重点を置いています。