Elasticsearch:MySQL から Alibaba Cloud Elasticsearch にデータを同期するために Canal を使用する

手順 1: MySQL データソースを準備する

ApsaraDB RDS コンソールにログインし、ApsaraDB RDS for MySQL データベースとテーブルを作成します。詳細については、ApsaraDB RDS for MySQL を使用する一般的なワークフローをご参照ください。この例では、次のステートメントを使用してテーブルを作成します。

-- create table  テーブルを作成する
CREATE TABLE `es_test` (
    `id` bigint(32) NOT NULL,
    `name` text NOT NULL,
    `count` text NOT NULL,
    `color` text NOT NULL,
    PRIMARY KEY (`id`)
) ENGINE=InnoDB
DEFAULT CHARACTER SET=utf8;

手順 2: インデックスを作成し、インデックスのマッピングを設定する

1. Elasticsearch クラスターの Kibana コンソールにログインします。詳細については、Kibana コンソールにログインするをご参照ください。

   説明

   この例では、Elasticsearch V6.7.0 クラスターを使用しています。他のバージョンのクラスターでの操作は異なる場合があります。

2. 表示されるページの左側のナビゲーションペインで、[dev Tools] をクリックします。

3. [console] タブで、次のコマンドを実行してインデックスを作成します。

   この例では、es_test という名前のインデックスが作成されます。インデックスには、count、id、name、color の各フィールドが含まれています。

   重要

   インデックスのマッピングで定義されているフィールドの名前とデータ型が、手順 1: MySQL データソースを準備するで準備したフィールドの名前とデータ型と同じであることを確認する必要があります。

   PUT es_test?include_type_name=true
   {
   
       "settings" : {
         "index" : {
           "number_of_shards" : "5",
           "number_of_replicas" : "1"
         }
       },
       "mappings" : {
           "_doc" : {
               "properties" : {
                 "count": {          // count フィールド
                      "type": "text"       // タイプを text に設定
                  },
                 "id": {          // id フィールド
                      "type": "integer"   // タイプを integer に設定
                  },
                  "name": {        // name フィールド
                       "type" : "text",   // タイプを text に設定
                       "analyzer": "ik_smart" // ik_smart アナライザーを使用                  
                   },
                   "color" : {      // color フィールド
                       "type" : "text"    // タイプを text に設定                   
                   }
               }
           }
       }
   }

   インデックスが作成され、マッピングが設定されると、次の結果が返されます。

   {
     "acknowledged" : true,
     "shards_acknowledged" : true,
     "index" : "es_test"
   }

手順 3: JDK をインストールする

1. ECS インスタンスに接続します。

   詳細については、パスワードまたはキーを使用して Linux インスタンスに接続するをご参照ください。

   説明

   この例では、通常のユーザーを使用しています。

2. 使用可能な JDK パッケージを表示します。

   sudo yum search java | grep -i --color JDK

3. 必要なバージョンの JDK をインストールします。

   この例では、[java-1.8.0-openjdk-devel.x86_64] を使用しています。

   sudo yum install java-1.8.0-openjdk-devel.x86_64

4. 環境変数を設定します。

   1. etc フォルダーにある profile ファイルを開きます。

      vim ~/.bash_profile

   2. 次の環境変数をファイルに追加します。

      export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.362.b08-1.el7_9.x86_64
      export CLASSPATH=.:$JAVA_HOME/jre/lib/rt.jar:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar
      export PATH=$PATH:$JAVA_HOME/bin

      重要

      環境変数を使用する場合は、JAVA_HOME を JDK のインストールパスに置き換える必要があります。find / -name 'java' コマンドを実行して、JDK のインストールパスを照会できます。

   3. [esc] を押し、:wq コマンドを実行してファイルを保存し、vi モードを終了します。次に、次のコマンドを実行して設定を適用します。

      source ~/.bash_profile

5. 次のコマンドを実行して、JDK がインストールされているかどうかを確認します。

   java -version

   次の結果が返された場合、JDK はインストールされています。

   openjdk version "1.8.0_362"
   OpenJDK Runtime Environment (build 1.8.0_362-b08)
   OpenJDK 64-Bit Server VM (build 25.362-b08, mixed mode)

手順 4: Canal サーバーをインストールして起動する

1. Canal サーバーパッケージをダウンロードします。

   この例では、Canal 1.1.4 サーバーを使用しています。

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

   説明

   • Canal 1.1.5 は Elasticsearch V7.0 クラスターをサポートしています。 Elasticsearch V7.0 クラスターを使用している場合は、Canal 1.1.5 パッケージをダウンロードする必要があります。詳細については、Canal リリースノートをご参照ください。

   • Canal サーバーと Canal アダプターのパッケージはインターネット経由でダウンロードする必要があります。 ECS インスタンスがインターネットにアクセスできることを確認してください。

2. 次のコマンドを実行してパッケージを解凍します。

   tar -zxvf canal.deployer-1.1.4.tar.gz

3. 次のコマンドを実行して、conf/example/ ディレクトリにある instance.properties ファイルを変更します。

   vi conf/example/instance.properties

   

   パラメーター	説明
   canal.instance.master.address	このパラメーターは、<ApsaraDB RDS for MySQL インスタンスの内部エンドポイント>:<内部ポート> の形式の値に設定する必要があります。必要な情報は、ApsaraDB RDS for MySQL インスタンスの [基本情報] ページで取得できます。例: rm-bp1u1xxxxxxxxx6ph.mysql.rds.aliyuncs.com:3306。
   canal.instance.dbUsername	ApsaraDB RDS for MySQL データベースにログインするために使用されるユーザー名。ユーザー名は、ApsaraDB RDS for MySQL インスタンスの [アカウント] ページで取得できます。
   canal.instance.dbPassword	ApsaraDB RDS for MySQL データベースにログインするために使用されるパスワード。

4. [esc] を押し、:wq コマンドを実行してファイルを保存し、vi モードを終了します。

5. Canal サーバーを起動し、ログをクエリします。

   ./bin/startup.sh
   cat logs/canal/canal.log

   

手順 5: Canal アダプターをインストールして起動する

1. Canal アダプターパッケージをダウンロードします。

   この例では、Canal 1.1.4 アダプターを使用しています。

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

   説明

   • Canal 1.1.5 は Elasticsearch V7.0 クラスタをサポートしています。Elasticsearch V7.0 クラスタを使用している場合は、Canal 1.1.5 パッケージをダウンロードする必要があります。詳細については、Canal リリースノート をご参照ください。

   • Canal サーバーと Canal アダプターのパッケージはインターネット経由でダウンロードする必要があります。 ECS インスタンスがインターネットにアクセスできることを確認してください。

2. 次のコマンドを実行してパッケージを解凍します。

   tar -zxvf canal.adapter-1.1.4.tar.gz

3. 次のコマンドを実行して、conf/ ディレクトリにある application.yml ファイルを変更します。

   vi conf/application.yml

   

   パラメーター	説明
   canal.conf.canalServerHost	Canal デプロイヤのアドレス。デフォルト値 127.0.0.1:11111 を保持します。
   canal.conf.srcDataSources.defaultDS.url	このパラメーターは、jdbc:mysql://<ApsaraDB RDS for MySQL インスタンスの内部エンドポイント>:<内部ポート>/<データベース名>?useUnicode=true の形式の値に設定する必要があります。必要な情報は、ApsaraDB RDS for MySQL インスタンスの [基本情報] ページで取得できます。例: jdbc:mysql://rm-bp1xxxxxxxxxnd6ph.mysql.rds.aliyuncs.com:3306/elasticsearch?useUnicode=true。
   canal.conf.srcDataSources.defaultDS.username	ApsaraDB RDS for MySQL データベースにログインするために使用されるユーザー名。ユーザー名は、ApsaraDB RDS for MySQL インスタンスの [アカウント] ページで取得できます。
   canal.conf.srcDataSources.defaultDS.password	ApsaraDB RDS for MySQL データベースにログインするために使用されるパスワード。
   canal.conf.canalAdapters.groups.outerAdapters.hosts	name:es を見つけ、hosts を <Elasticsearch クラスターの内部エンドポイント>:<内部ポート> の形式の値に設定します。必要な情報は、Elasticsearch クラスターの [基本情報] ページで取得できます。例: es-cn-v64xxxxxxxxx3medp.elasticsearch.aliyuncs.com:9200。
   canal.conf.canalAdapters.groups.outerAdapters.mode	このパラメーターを rest に設定します。
   canal.conf.canalAdapters.groups.outerAdapters.properties.security.auth	このパラメーターは、<Elasticsearch クラスター>:<ユーザー名に対応するパスワード> の形式の値に設定する必要があります。例: elastic:es_password。
   canal.conf.canalAdapters.groups.outerAdapters.properties.cluster.name	Elasticsearch クラスター。 ID は、Elasticsearch クラスターの [基本情報] ページで取得できます。例: es-cn-v64xxxxxxxxx3medp。

4. [esc] を押し、:wq コマンドを実行してファイルを保存し、vi モードを終了します。

5. 前の手順を繰り返して、conf/es/ ディレクトリにある *.yml ファイルを変更し、ApsaraDB RDS for MySQL データベースから Elasticsearch クラスターにマップするフィールドを指定します。

   

   パラメーター	説明
   esMapping._index	手順 2: インデックスを作成し、インデックスのマッピングを設定するで Elasticsearch クラスターに作成したインデックスの名前に値を設定します。この例では、[es_test] を使用しています。
   esMapping._type	手順 2: インデックスを作成し、インデックスのマッピングを設定するで Elasticsearch クラスターに作成したインデックスのタイプに値を設定します。この例では、[_doc] を使用しています。
   esMapping._id	Elasticsearch クラスターに同期するフィールドに対して生成されるドキュメントの ID。カスタム ID を指定できます。この例では、[_id] を使用しています。
   esMapping.sql	Elasticsearch クラスターに同期するフィールドをクエリするために使用される SQL ステートメント。この例では、select t.id as _id,t.id,t.count,t.name,t.color from es_test t ステートメントを使用しています。 // es_test テーブルからすべてのフィールドをクエリする

6. Canal アダプターを起動し、ログをクエリします。

   ./bin/startup.sh
   cat logs/adapter/adapter.log

   説明

   この例では、MySQL 5.7 を実行する ApsaraDB RDS for MySQL インスタンスを使用しています。別のバージョンの MySQL を実行する ApsaraDB RDS for MySQL インスタンスを使用する場合は、Canal アダプターの MySQL ドライバーのバージョンが、接続する ApsaraDB RDS for MySQL インスタンスの MySQL バージョンと一致していることを確認する必要があります。そうでない場合、Canal アダプターは起動に失敗します。詳細については、このトピックの FAQ セクションをご参照ください。

   次の図に示す結果が返された場合、Canal アダプターは起動されています。

手順 6: 増分データの同期結果を確認する

1. ApsaraDB RDS for MySQL データベースで、[es_test] テーブルにデータを追加、変更、または削除します。

   insert `ES`.`es_test`(`count`,`id`,`name`,`color`) values('11',2,'canal_test2','red');

2. Elasticsearch クラスターの Kibana コンソールにログインします。詳細については、Kibana コンソールにログインするをご参照ください。

3. 表示されるページの左側のナビゲーションペインで、[dev Tools] をクリックします。

4. [console] タブで、次のコマンドを実行して同期されたデータをクエリします。

   GET /es_test/_search

   データ同期が成功すると、次の図に示す結果が返されます。

   重要

   Canal は増分データのみ同期します。

Q: アダプターの起動時に、Canal アダプターに対して生成されたログに次のエラーメッセージが返される場合はどうすればよいですか: java.lang.RuntimeException: java.lang.RuntimeException: java.lang.ClassCastException: com.alibaba.druid.pool.DruidDataSource cannot be cast to com.alibaba.druid.pool.DruidDataSource

at com.alibaba.otter.canal.client.adapter.es7x.ES7xAdapter.init(ES7xAdapter.java:54) ~[client-adapter.es7x-1.1.5-jar-with-dependencies.jar:na]

A: canal.adapter-1.1.5\plugin ディレクトリにある client-adapter.es7x-1.1.5-jar-with-dependencies.jar を canal-1.1.5-alpha-2 の関連ファイルに置き換えます。

この例では、root ユーザーを使用しています。

1. canal-1.1.5-alpha-2 のパッケージをダウンロードします。詳細については、Canal リリースノート をご参照ください。

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

2. パッケージを解凍します。

   tar -zxvf canal.adapter-1.1.5-SNAPSHOT.tar.gz

3. canal-1.1.5-alpha-2 の plugin フォルダーにある client-adapter.es7x-1.1.5-SNAPSHOT-jar-with-dependencies.jar を canal.adapter-1.1.5\plugin ディレクトリにコピーします。

   説明

   コピーする必要があるファイルのディレクトリは異なる場合があります。実際のディレクトリが優先されます。

   cp canal.adapter-1.1.5-SNAPSHOT/plugin/client-adapter.es7x-1.1.5-SNAPSHOT-jar-with-dependencies.jar canal/canal.adapter/plugin

4. canal.adapter-1.1.5\plugin ディレクトリから client-adapter.es7x-1.1.5-jar-with-dependencies.jar を削除します。

   rm -rf client-adapter.es7x-1.1.5-jar-with-dependencies.jar

5. ファイル名を変更します。

   mv client-adapter.es7x-1.1.5-SNAPSHOT-jar-with-dependencies.jar client-adapter.es7x-1.1.5-jar-with-dependencies.jar 

Q: アダプターの起動時に、Canal アダプターに対して生成されたログに次のエラーメッセージが返される場合はどうすればよいですか: java.sql.SQLException: Unknown system variable 'query_cache_size'?

A: 考えられる原因は、Canal アダプターの MySQL ドライバーのバージョンが、接続する ApsaraDB RDS for MySQL データベースの MySQL バージョンと一致していないことです。たとえば、Canal アダプター 1.1.4 の MySQL ドライバーのバージョンは 5.1.40 です。このバージョンの Canal アダプターを使用する Canal を使用して、MySQL 8.x を実行する ApsaraDB RDS for MySQL データベースに接続すると、エラーメッセージが返されます。 Canal アダプターの MySQL ドライバーを変更し、MySQL ドライバーのバージョンが ApsaraDB RDS for MySQL データベースの MySQL バージョンと一致していることを確認してください。

Q: MySQL 8.0 を実行する ApsaraDB RDS for MySQL インスタンスから Elasticsearch クラスターにデータを同期したいと考えています。既存の MySQL ドライバーを MySQL 8.0 ドライバーに変更するにはどうすればよいですか?

A: この例では、root ユーザーを使用しています。

1. MySQL 8.0 ドライバーのパッケージをダウンロードします。

   wget https://dev.mysql.com/get/Downloads/Connector-J/mysql-connector-java-8.0.29.zip

2. パッケージを解凍します。

   unzip mysql-connector-java-8.0.29.zip

3. 取得したファイルを Canal アダプターの lib ディレクトリにコピーします。

   mv mysql-connector-java-8.0.29/mysql-connector-java-8.0.29.jar lib/

4. 権限を追加します。

   chmod 777 lib/mysql-connector-java-8.0.29.jar
   chmod +st lib/mysql-connector-java-8.0.29.jar

5. MySQL 5.x ドライバーを削除します。

   rm -rf lib/mysql-connector-java-5.1.40.jar