Canalを使用してMySQLデータベースからOceanBaseデータベースへデータを同期する

CanalはAlibabaがオープンソースで公開した製品で、主にMySQLデータベースの増分ログを解析し、増分データのサブスクリプションと消費を提供します。

このドキュメントでは、主にCanalのcanal.deployerおよびcanal.adapterコンポーネントを使用して、MySQLデータベースからOceanBaseデータベースへデータを同期する方法について説明します。

アーキテクチャの原理

• canal deployer：canal の server 側で、binlog を CanalEntry に変換します。

• canal adapter：canal のクライアントアダプターで、CanalEntry を解析し、増分変更を宛先に同期します。

Canal関連情報アクセスアドレス：https://github.com/alibaba/canal/releases。

操作手順

ステップ１：MySQL関連設定

1. MySQL設定ファイル my.cnf を修正します。

   設定ファイル my.cnf の場所：/etc/my.cnf。まずbinlog書き込み機能を有効にし、binlog-formatをROWモードに設定します。設定後はMySQLを再起動して有効にする必要があります。

   例：

   log-bin=mysql-bin # binlogを有効にする
   binlog-format=ROW # ROWモードを選択する
   server_id=1 # MySQLレプリケーションで定義が必要なパラメータで、canalのslaveIdと重複しないようにする
   

2. MySQLユーザーを作成します。

   MySQLに接続するユーザーを作成します。ユーザー名：canal、パスワード：******。そして、canal にすべてのデータベースの読み書き権限を付与します。

   例：

   MySQL [(none)]> CREATE USER 'canal'@'%' IDENTIFIED BY '******';
   Query OK, 0 rows affected
   
   MySQL [(none)]> GRANT SELECT,REPLICATION SLAVE,REPLICATION CLIENT ON *.* TO 'canal'@'%';
   Query OK, 0 rows affected
   
   MySQL [(none)]> FLUSH PRIVILEGES;
   Query OK, 0 rows affected
   

3. テストデータベースを作成します。

   データベース test_mysql_to_ob、テーブル tbl1 および tbl2 を作成し、データを挿入します。

   MySQL [(none)]> CREATE DATABASE test_mysql_to_ob;
   Query OK, 1 row affected
   
   MySQL [(none)]> USE test_mysql_to_ob;
   Database changed
   MySQL [test_mysql_to_ob]> CREATE TABLE tbl1(col1 INT PRIMARY KEY, col2 VARCHAR(20),col3 INT);
   Query OK, 0 rows affected
   
   MySQL [test_mysql_to_ob]> INSERT INTO tbl1 VALUES(1,'China',86),(2,'Taiwan',886),(3,'Hong Kong',852),(4,'Macao',853),(5,'North Korea',850);
   Query OK, 5 rows affected
   Records: 5  Duplicates: 0  Warnings: 0
   
   MySQL [test_mysql_to_ob]> CREATE TABLE tbl2(col1 INT PRIMARY KEY,col2 VARCHAR(20));
   Query OK, 0 rows affected
   
   MySQL [test_mysql_to_ob]> INSERT INTO tbl2 VALUES(86,'+86'),(886,'+886'),(852,'+852'),(853,'+853'),(850,'+850');
   Query OK, 5 rows affected
   Records: 5  Duplicates: 0  Warnings: 0
   

ステップ2：Canalのダウンロードとインストール

1. ソフトウェアパッケージをダウンロードします。

   canal.deployer-1.1.5.tar.gz をダウンロードします。

   wget https://github.com/alibaba/canal/releases/download/canal-1.1.5/canal.deployer-1.1.5.tar.gz
   

2. 圧縮ファイルを /Canal_Home/canal ディレクトリに解凍します。

   mkdir /Canal_Home/canal && tar zxvf canal.deployer-1.1.5.tar.gz  -C /Canal_Home/canal
   

3. 設定ファイルを変更します。

   canal.deployer のデフォルト設定ファイルは conf/canal.properties と conf/example/instance.properties です。デフォルトでは example という名前の instance が作成されます。example インスタンスの設定ファイルを変更し、データベース接続アドレス、ユーザー名、パスワードを修正する必要があります。canal.instance.connectionCharsets は、データベースのエンコーディング方式をJavaのエンコーディングタイプに対応付けるもので、例えば UTF-8、GBK、ISO-8859-1 などが該当します。

   例：

   vi conf/example/instance.properties
   
   # mysql serverId
   canal.instance.mysql.slaveId = 1234
   #position info、ご自身のデータベース情報に変更する必要があります
   canal.instance.master.address = xxx.xxx.xxx.xxx:3306
   canal.instance.master.journal.name =
   canal.instance.master.position =
   canal.instance.master.timestamp =
   #canal.instance.standby.address =
   #canal.instance.standby.journal.name =
   #canal.instance.standby.position =
   #canal.instance.standby.timestamp =
   #username/password、ご自身のデータベース情報に変更する必要があります
   canal.instance.dbUsername = canal
   canal.instance.dbPassword = ******
   canal.instance.defaultDatabaseName =
   canal.instance.connectionCharsets = UTF-8
   #table regex
   canal.instance.filter.regex = .*\\..*
   

4. Canal Serverを起動します。

   cd /Canal_Home/canal && sh bin/startup.sh
   

5. サーバーログを確認します。

   cat logs/canal/canal.log
   

6. インスタンスのログを確認します。

   tail -f logs/canal/canal.log
   tail -f logs/example/example.log
   

7. サービスを停止する場合は、以下のコマンドを実行します。

   cd /Canal_Home/canal && sh bin/stop.sh
   

ステップ3：RDBアダプターのデプロイ

Canal Adapterは多様なターゲットコンテナをサポートしていますが、OceanBaseでは主にそのrdbモジュールを使用し、ターゲットコンテナはOceanBase自身となります。

1. ソフトウェアパッケージをダウンロードします。

   canal(adapter-1.1.5.tar.gzをダウンロードします。

   wget https://github.com/alibaba/canal/releases/download/canal-1.1.5/canal.adapter-1.1.5.tar.gz
   

2. 圧縮ファイルを/Canal_Home/adapterディレクトリに解凍します。

   mkdir /Canal_Home/adapter && tar zxvf canal.adapter-1.1.5.tar.gz  -C /Canal_Home/adapter
   

3. ローダーの設定を変更します。

   ローダーの設定ファイルconf/application.ymlを変更します。まずadapterのデータ送信元タイプを指定し、modeで指定します。ここではtcpを選択します。その後、canal.tcp関連のプロパティを指定します。これにはcanal serverのIPアドレスとポート、データベースの接続ユーザー名とパスワードが含まれます。次にadapterのデータ送信先接続情報を指定します。instanceはデータ送信元インスタンス名で、canalデプロイ時に定義されます。keyはカスタムキーで、後の処理で使用されます。jdbc関連のプロパティはデータ送信先OceanBaseへの接続方式で、MySQL付属のドライバーを使用できます。

   例：

   mode: tcp #tcp kafka rocketMQ rabbitMQ
   flatMessage: true
   zookeeperHosts:
   syncBatchSize: 1000
   retries: 0
   timeout:
   accessKey:
   secretKey:
   consumerProperties:
   # canal tcp consumer
       canal.tcp.server.host: 127.0.0.1:11111
       canal.tcp.zookeeper.hosts:
       canal.tcp.batch.size: 500
       canal.tcp.username:
       canal.tcp.password:
   canalAdapters:
   - instance: example # canal instance Name or mq topic name
     groups:
     - groupId: g1
       outerAdapters:
       - name: logger
       - name: rdb
         key: test_mysql_to_ob
         properties:
           jdbc.driverClassName: com.mysql.jdbc.Driver
           jdbc.url: jdbc:mysql://xxx.xxx.xxx.xxx:2883/test_data?useUnicode=true
           jdbc.username: root@mysql001#test4000
           jdbc.password: ******
   

4. RDBマッピングファイル。

   conf/rdb/mytest_user.ymlファイルを変更します。ここで、destinationはcanal instance名を指定します；outerAdapterKeyは前述で定義したkeyです；mirrorDbはデータベースレベルのDDLおよびDMLイメージ同期を指定します。

   マッピングには2種類あります：1つはテーブル単位のマッピング、もう1つはデータベース全体のマッピングです。ここではデータベース全体のマッピングを例に設定します。コメント部分はテーブル単位のマッピングの設定です：

   [root@obce00 adapter]# cat conf/rdb/mytest_user.yml
   #dataSourceKey: defaultDS
   #destination: example
   #groupId: g1
   #outerAdapterKey: mysql1
   #concurrent: true
   #dbMapping:
   #  database: mytest
   #  table: user
   #  targetTable: mytest
   #  targetPk:
   #    id: id
   #  mapAll: true
   #  targetColumns:
   #    id:
   #    name:
   #    role_id:
   #    c_time:
   #    test1:
   #  etlCondition: "where c_time>={}"
   #  commitBatch: 3000 # バッチ送信のサイズ
   # Mirror schema synchronize config
   dataSourceKey: defaultDS
   destination: example
   groupId: g1
   outerAdapterKey: test_mysql_to_ob
   concurrent: true
   dbMapping:
   mirrorDb: true
   database: test_data
   commitBatch: 1000
   

5. RDBを起動します。

canal-adapterローダーを起動します。

```bash
cd /Canal_Home/adapter && sh bin/startup.sh
```

6. RDBログを確認します。

   tail -f logs/adapter/adapter.log
   

7. サービスを停止するには、以下のコマンドを実行します。

   cd /Canal_Home/adapter && bin/stop.sh
   

8. データ同期状況を確認します。

   MySQLのデータ送信元にデータを書き込み、OceanBaseのデータ送信先でデータ同期を確認します。

機能制限

• 同期するテーブルにはプライマリキーが必要です。プライマリキーのないテーブルをデータ送信元で削除すると、データ送信先に同期された際にテーブル全体が削除されます。

• DDLでは、テーブルの新規作成と列の追加をサポートします。