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

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

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

アーキテクチャの原理

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

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

Canal関連情報のアクセスURL：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 replactionの設定には定義が必要であり、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種類あります。一つはテーブル単位のマッピング、もう一つはデータベース全体のマッピングです。以下はデータベース全体のマッピングを例にした設定です。コメント部分はテーブル単位のマッピングの設定です：

   [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では、新規テーブルの作成や列の追加がサポートされています。