このトピックでは、Canal を使用して、ApsaraDB RDS for MySQL の増分データを Alibaba Cloud Elasticsearch と同期させる方法について説明します。

始める前に

Canal を使用する前に、次の前提条件を満たしていることを確認してください。
重要 ApsaraDB RDS for MySQL、Alibaba Cloud Elasticsearch、Elastic Compute Service (ECS) を有効化するとき、これらのサービスに同じリージョンゾーンVPC ネットワークVSwitchセキュリティ グループが指定されていることを確認します。
  • ApsaraDB RDS for MySQL
    ApsaraDB RDS for MySQL を使用して、同期対象のソースデータと増分データを保存します。 ApsaraDB RDS for MySQL を有効化する方法の詳細は、「インスタンスの作成」をご参照ください。 次の図に、このトピックで使用する ApsaraDB RDS for MySQL の設定を示します。ApsaraDB RDS for MySQL 設定
  • canal.adapter-1.1.4.tar.gz とcanal.deployer-1.1.4.tar.gz

    Canal パッケージです。 Canal は、GitHub オープンソースの extract-transform-load (ETL) ツールです。このツールを使用してデータベースログを解析し、同期対象の増分データを取得します。 詳細は、「Canal」をご参照ください。

  • Alibaba Cloud Elasticsearch
    Alibaba Cloud Elasticsearch を使用して、同期対象の増分データを受信します。 Alibaba Cloud Elasticsearch を有効化する方法の詳細は、「Alibaba Cloud Elasticsearch インスタンスの作成」をご参照ください。 次の図に、このトピックで使用する Elasticsearch の設定を示します。Elasticsearch 設定
  • Alibaba Cloud ECS
    Alibaba Cloud ECS を使用して、ApsaraDB RDS for MySQL と Alibaba Cloud Elasticsearch を接続します。 また、Canal deployer と Canal adapter を Alibaba Cloud ECS にデプロイします。 Alibaba Cloud ECS を有効化する方法の詳細は、「手順 2: インスタンスの作成」をご参照ください。 次の図に、このトピックで使用する ECS の設定を示します。ECS 設定

テーブル、フィールド、インデックスの作成

  1. ApsaraDB RDS for MySQL にテーブルを作成し、このテーブルにフィールドを追加します。
    このトピックでは、テーブル es_test を作成します。 次の図に、テーブルに追加するフィールドを示します。 es_test テーブルとフィールド
  2. Elasticsearch インスタンスにインデックスを作成し、マッピングを設定します。
    Kibana コンソールへのログインを実行し、[Dev Tools] ページ の [Console] で次のコマンドを実行してインデックスを作成し、マッピングを設定します。
    重要 次のコマンドで指定されているテーブル名、フィールド名、フィールドタイプは、MySQL テーブルのテーブル名、フィールド名、フィールドタイプと同じでなければなりません。
    PUT es_test? include_type_name=true
    {
    
        "settings" : {
          "index" : {
            "number_of_shards" : "5",
            "number_of_replicas" : "1"
          }
        },
        "mappings" : {
            "_doc" : {
                "properties" : {
                  "count": {          
                       "type": "text"       
                   },
                  "id": {
                       "type": "long"
                   },
                    "name" : {
                        "type" : "text"                    
                    },
                    "color" : {
                        "type" : "text"                    
                    }
                }
            }
        }
    }
    テーブルとマッピングが正常に作成されると、次の結果が返されます。
    {
      "acknowledged" : true,
      "shards_acknowledged" : true,
      "index" : "es_test"
    }

MySQL のインストール

  1. Alibaba Cloud ECS インスタンスに接続します。
  2. MySQL ソースインストールパッケージをダウンロードします。
    wget http://dev.mysql.com/get/mysql57-community-release-el7-11.noarch.rpm
  3. MySQL ソースをインストールします。
    yum -y install mysql57-community-release-el7-11.noarch.rpm
  4. MySQL ソースが正常にインストールされたかどうかを確認します。
    yum repolist enabled | grep mysql. *
    MySQL ソースが正常にインストールされると、次の結果が返されます。MySQL ソースのインストール成功
  5. MySQL サーバーをインストールします。
    yum install mysql-community-server
  6. MySQL サービスを起動し、MySQL サービスのステータスを確認します。
    systemctl start mysqld.service
    systemctl status mysqld.service
    MySQL サービスが起動されていると、次の結果が返されます。MySQL の起動とステータスの確認
  7. ApsaraDB RDS for MySQL データベースに接続します。
    重要
    • 次のコマンドを実行して ApsaraDB RDS for MySQL データベースに接続する前に、ECS インスタンスのプライベート IP アドレスを ApsaraDB RDS for MySQL ホワイトリストに追加する必要があります。 詳細は、「ホワイトリストの設定」をご参照ください。
    • Canal を使用するには、MySQL バイナリログモードを有効にする必要があります。 ApsaraDB RDS for MySQL の場合、このモードはデフォルトで有効になっています。 次のコマンドを実行すると、バイナリログモードのステータスを照会できます。
      show variables like '%log_bin%';
      バイナリログモードが有効になっている場合、次の結果が返されます。MySQL バイナリログモードの有効化
    mysql -h<hostname> -P<port> -u<username> -p<password> -D<database>
    変数 説明
    <hostname> ApsaraDB RDS for MySQL インスタンスのイントラネットアドレス。 イントラネットアドレスの情報は、インスタンスの [基本情報] で確認できます。
    <port> ApsaraDB RDS for MySQL インスタンスの内部ポート。 デフォルトのポート番号は、3306 です。 内部ポートの情報は、インスタンスの [基本情報] ページで確認できます。
    <username> ApsaraDB RDS for MySQL データベースのユーザー名。 アカウント情報は、インスタンスの [アカウント管理] ページで確認できます。 使用可能なアカウントがない場合、アカウントを作成する必要があります。 詳細は、「アカウントとデータベースの作成」をご参照ください。
    <database> ApsaraDB RDS for MySQL データベースの名前。 データベース名は、インスタンスの [データベース管理] ページで確認できます。 使用可能なデータベースがない場合、データベースを作成する必要があります。 詳細は、「アカウントとデータベースの作成」をご参照ください。
    <password> ApsaraDB RDS for MySQL データベースのパスワード。
    コマンド例:
    mysql -hrm-bp1u1xxxxxxxxx6ph.mysql.rds.aliyuncs.com -P3306 -ues -pmima -Delasticsearch
    ApsaraDB RDS for MySQL に接続されていると、次の結果が返されます。ApsaraDB RDS for MySQL への接続

JDK のインストール

  1. ECS インスタンスに接続し、使用可能な JDK パッケージを照会します。
    yum search java | grep -i --color JDK
  2. バージョンを選択し、JDK をインストールします。 このトピックでは、java-1.8.0-openjdk-devel.x86_64 を選択しています。
    yum install java-1.8.0-openjdk-devel.x86_64
  3. 環境変数を設定します。
    1. etc フォルダーの profile ファイルを開きます。
      vi /etc/profile
    2. 次の環境変数をファイルに追加します。
      export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.71-2.b15.el7_2.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
    3. :wq と入力してファイルを保存し、vi モードを終了します。続いて、次のコマンドを実行して変更を適用します。
      source /etc/profile
  4. 次のコマンドを実行して、JDK が正常にインストールされたかどうかを確認します。
    • java
    • javac
    • java -version
      JDK が正常にインストールされていると、次の結果が返されます。JDK のインストール成功

Canal サーバーのインストールと起動

  1. ECS インスタンスに接続し、Canal deployer パッケージをダウンロードして解凍します。 このトピックでは、Canal deployer V1.1.4 を使用しています。
    wget https://github.com/alibaba/canal/releases/download/canal-1.1.4/canal.deployer-1.1.4.tar.gz
  2. canal.deployer-1.1.4.tar.gz パッケージを解凍します。
    tar -zxvf canal.deployer-1.1.4.tar.gz
  3. conf/example/instance.properties ファイルを変更します。
    vi conf/example/instance.properties
    conf/example/instance.property ファイルの変更
    パラメーター 説明
    canal.instance.master.address <ApsaraDB RDS for MySQL instance internal endpoint>:<internal port>。 必要な情報は、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. :wq と入力してファイルを保存し、vi モードを終了します。
  5. Canal サーバーを起動し、Canal ログを表示します。
    ./bin/startup.sh
    cat logs/canal/canal.log
    Canal サーバーの起動

Canal adapter のインストールと起動

  1. ECS インスタンスに接続し、Canal adapter パッケージをダウンロードして解凍します。 このトピックでは、Canal adapter V1.1.4 を使用しています。
    wget https://github.com/alibaba/canal/releases/download/canal-1.1.4/canal.adapter-1.1.4.tar.gz
  2. canal.adapter-1.1.4.tar.gz パッケージを解凍します。
    tar -zxvf canal.adapter-1.1.4.tar.gz
  3. conf/application.yml ファイルを変更します。
    vi conf/application.yml
    conf/application.yml ファイルの変更
    パラメーター 説明
    canal.conf.canalServerHost Canal deployer のエンドポイント。 デフォルトの設定 127.0.0.1:11111 を使用します。
    canal.conf.srcDataSources.defaultDS.url jdbc:mysql://<ApsaraDB RDS for MySQL instance internal endpoint>:<internal port>/<database name>? 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<Alibaba Cloud Elasticsearch instance internal network address>:<internal network port> に置き換えます。 必要な情報は、Elasticsearch インスタンスの [概要 ] ページで確認できます。 例:es-cn-v64xxxxxxxxx3medp.elasticsearch.aliyuncs.com:9200
    canal.conf.canalAdapters.groups.outerAdapters.mode 値を rest に設定します。
    canal.conf.canalAdapters.groups.outerAdapters.properties.security.auth 値を <Alibaba Cloud Elasticsearch instance username>:<password> に設定します。 例:elastic:es_password
    canal.conf.canalAdapters.groups.outerAdapters.properties.cluster.name Alibaba Cloud Elasticsearch インスタンスの ID。 ID は、Elasticsearch インスタンスの [概要 ] ページで確認できます。 例:es-cn-v64xxxxxxxxx3medp
  4. :wq と入力してファイルを保存し、vi モードを終了します。
  5. 同じ手順で conf/es/*.yml ファイルを変更し、ApsaraDB RDS for MySQL から Alibaba Cloud Elasticsearch にマッピングするフィールドを指定します。
    conf/es/*.yml ファイルの変更
    パラメーター 説明
    esMapping._index この値には、「テーブル、フィールド、インデックスの作成」で Elasticsearch インスタンスに作成したインデックス名を設定します。 このトピックでは、値に es_test を設定しています。
    esMapping._type この値には、「テーブル、フィールド、インデックスの作成」で 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; 文を使用しています。
  6. Canal adapter サービスを起動し、Canal ログを表示します。
    ./bin/startup.sh
    cat logs/adapter/adapter.log
    Canal adapter サービスが起動されていると、次の結果が返されます。Canal adapter サービスログ

Canal の増分データ同期機能の検証

  1. ApsaraDB RDS for MySQL データベースのテーブル es_test に対して、データの追加、変更、または削除を行います。
    insert `elasticsearch`.`es_test`(`count`,`id`,`name`,`color`) values('11',2,'canal_test2','red');
  2. Alibaba Cloud Elasticsearch コンソールにログインし、Kibana コンソールへのログインを実行します。
  3. Kibana コンソールの [Dev Tools] ページの [Console] で、同期対象の増分データを照会します。
    GET /es_test/_search
    増分データが正常に同期されていると、次の結果が返されます。データ同期が成功した結果