Elasticsearch:Logstash を使用して ApsaraDB RDS for MySQL から Elasticsearch にデータを同期する

背景情報

Alibaba Cloud Logstash は、データの収集、変換、最適化、および生成に使用できる強力なデータ処理ツールです。 Alibaba Cloud Logstash は logstash-input-jdbc プラグインを提供します。このプラグインはデフォルトで Logstash クラスタにインストールされており、削除することはできません。 このプラグインは、ApsaraDB RDS for MySQL の複数のデータレコードを一度にクエリし、それらのレコードを Alibaba Cloud Elasticsearch に同期できます。 このプラグインはラウンドロビン方式を使用して、ApsaraDB RDS for MySQL に最後に挿入または更新されたデータレコードを定期的に識別し、それらのレコードを Alibaba Cloud Elasticsearch に同期します。 詳細については、「Logstash と JDBC を使用してリレーショナルデータベースから Elasticsearch クラスタにデータを同期する」をご参照ください。 完全データを同期し、数秒の遅延を許容できる場合、または一度に特定のデータをクエリして同期する場合は、Alibaba Cloud Logstash を使用できます。

前提条件

同じ仮想プライベートクラウド（VPC）内に次のインスタンスとクラスタを作成することをお勧めします。

• ApsaraDB RDS for MySQL インスタンスを作成します。 詳細については、「ApsaraDB RDS for MySQL インスタンスを作成する」をご参照ください。 この例では、MySQL 5.7 を実行する ApsaraDB RDS インスタンスが作成されます。

• Alibaba Cloud Elasticsearch クラスタを作成します。 詳細については、「Alibaba Cloud Elasticsearch クラスタを作成する」をご参照ください。 この例では、Elasticsearch V7.10 クラスタが作成されます。

• Alibaba Cloud Logstash クラスタを作成します。 詳細については、「Alibaba Cloud Logstash クラスタを作成する」をご参照ください。

制限事項

手順

FAQ

• Q：パイプラインが初期化状態でスタックしている、同期前後のデータに矛盾がある、またはデータベースへの接続に失敗したためにデータ同期タスクが失敗した場合はどうすればよいですか？

  A：Logstash クラスタのクラスタログにエラー情報が含まれているかどうかを確認し、エラー情報に基づいて原因を特定します。 詳細については、「ログをクエリする」をご参照ください。 次の表に、エラーの一般的な原因とエラーの解決策を示します。

  説明

  次の解決策で説明されている操作を実行するときに Logstash クラスタで更新操作が実行されている場合は、「クラスタタスクの進捗状況を表示する」を参照して更新操作を一時停止します。 解決策で説明されている操作が完了すると、システムは Logstash クラスタを再起動し、更新操作を再開します。

  原因	解決策
  Logstash クラスタのノードの IP アドレスが ApsaraDB RDS for MySQL インスタンスのホワイトリストに追加されていません。	Logstash クラスタのノードの IP アドレスを ApsaraDB RDS for MySQL インスタンスのホワイトリストに追加します。 詳細については、「データベースクライアントまたは CLI を使用して ApsaraDB RDS for MySQL インスタンスに接続する」をご参照ください。 説明 Logstash クラスタのノードの IP アドレスを取得する方法の詳細については、「クラスタの基本情報を表示する」をご参照ください。
  ECS インスタンスでホストされているセルフマネージド MySQL データベースからデータを同期するために Logstash クラスタを使用していますが、クラスタのノードのプライベート IP アドレスと内部ポートが ECS インスタンスのセキュリティグループに追加されていません。	クラスタのノードのプライベート IP アドレスと内部ポートを ECS インスタンスのセキュリティグループに追加します。 詳細については、「セキュリティグループルールを追加する」をご参照ください。 説明 Logstash クラスタのノードの IP アドレスとポートを取得する方法の詳細については、「クラスタの基本情報を表示する」をご参照ください。
  Elasticsearch クラスタが Logstash クラスタと同じ VPC にありません。	次のいずれかの解決策を使用します。 Logstash クラスタと同じ VPC にある Elasticsearch クラスタを購入します。 詳細については、「Alibaba Cloud Elasticsearch クラスタを作成する」をご参照ください。 Elasticsearch クラスタを購入した後、Logstash クラスタのパイプライン設定を変更します。 インターネット経由でデータを送信するように NAT ゲートウェイを構成します。 詳細については、「インターネット経由のデータ転送用の NAT ゲートウェイを構成する」をご参照ください。
  ApsaraDB RDS for MySQL インスタンスのエンドポイントが正しくないか、インスタンスのポート番号が 3306 ではありません。	正しいエンドポイントとポート番号を取得します。 詳細については、「インスタンスのエンドポイントとポートを表示および管理する」をご参照ください。 次に、jdbc_connection_string パラメーターの値のエンドポイントとポート番号を取得したエンドポイントとポート番号に置き換えます。 重要 <ApsaraDB RDS for MySQL インスタンスのエンドポイント>：ApsaraDB RDS for MySQL インスタンスの内部エンドポイントを指定する必要があります。 ApsaraDB RDS for MySQL インスタンスのパブリックエンドポイントを使用する場合は、Logstash クラスタがインターネットに接続できるように、Logstash クラスタの NAT ゲートウェイを構成する必要があります。 詳細については、「インターネット経由のデータ転送用の NAT ゲートウェイを構成する」をご参照ください。
  Elasticsearch クラスタで自動インデックス作成機能が無効になっています。	Elasticsearch クラスタの自動インデックス作成機能を有効にします。 詳細については、「YML ファイルを構成する」をご参照ください。
  Elasticsearch クラスタまたは Logstash クラスタの負荷が過度に高くなっています。	Elasticsearch クラスタまたは Logstash クラスタの構成をアップグレードします。 詳細については、「クラスタの構成をアップグレードする」をご参照ください。 説明 Elasticsearch クラスタの負荷が過度に高い場合は、Elasticsearch コンソールでメトリックに基づいて収集された監視データを表示して、Elasticsearch クラスタの負荷情報を取得できます。 詳細については、「メトリックと例外処理の推奨事項」をご参照ください。 Logstash クラスタの負荷が過度に高い場合は、Logstash クラスタの X-Pack Monitoring 機能を有効にして、この機能を使用して Logstash クラスタを監視し、監視データを表示できます。 詳細については、「X-Pack Monitoring 機能を有効にする」をご参照ください。
  JDBC ベースの接続に ApsaraDB RDS for MySQL データベースに使用されるドライバファイルがアップロードされていません。	ドライバファイルをアップロードします。 詳細については、「サードパーティライブラリを構成する」をご参照ください。
  file_extend がパイプラインの構成で指定されています。 ただし、logstash-output-file_extend プラグインがインストールされていません。	次のいずれかの解決策を使用します。 logstash-output-file_extend プラグインをインストールします。 詳細については、「プラグインをインストールおよび削除する」をご参照ください。 パイプラインの構成から file_extend パラメーターを削除します。

  この問題の原因と解決策の詳細については、「Logstash を使用したデータ転送に関する FAQ」をご参照ください。

• Q：パイプラインの入力設定でソースデータベースへの複数の JDBC 接続を構成するにはどうすればよいですか？

  A：パイプラインの入力設定で複数の JDBC データソースを定義し、statement パラメーターで各テーブルのクエリステートメントを指定できます。 次のコードは例を示しています。

  input {
      jdbc {
        jdbc_driver_class => "com.mysql.jdbc.Driver" // JDBC ドライバーのクラス。
        jdbc_driver_library => "/ssd/1/share/<Logstash cluster ID>/logstash/current/config/custom/mysql-connector-java-5.1.48.jar" // JDBC ドライバーライブラリへのパス。
        jdbc_connection_string => "jdbc:mysql://rm-bp1xxxxx.mysql.rds.aliyuncs.com:3306/<Database name>?useUnicode=true&characterEncoding=utf-8&useSSL=false&allowLoadLocalInfile=false&autoDeserialize=false" // MySQL データベースへの JDBC 接続文字列。
        jdbc_user => "xxxxx" // MySQL データベースユーザー名。
        jdbc_password => "xxxx" // MySQL データベースパスワード。
        jdbc_paging_enabled => "true" // ページングが有効になっているかどうか。
        jdbc_page_size => "50000" // ページサイズ。
        statement => "select * from tableA where update_time >= :sql_last_value" // データをクエリするための SQL ステートメント。
        schedule => "* * * * *" // SQL ステートメントが実行される頻度。
        record_last_run => true // 最後の実行結果が記録されるかどうか。
        last_run_metadata_path => "/ssd/1/<Logstash cluster ID>/logstash/data/last_run_metadata_update_time.txt" // 最後の実行時間を含むファイルへのパス。
        clean_run => false // last_run_metadata_path パラメーターで指定されたパスがクリアされるかどうか。
        tracking_column_type => "timestamp" // 追跡されるカラムのタイプ。
        use_column_value => true // 特定のカラムの値が記録されるかどうか。
        tracking_column => "update_time" // 追跡されるカラム。
        type => "A" // データソースを識別するためのタイプ。
      }
      jdbc {
        // tableB の設定。tableA の設定に似ています。
        type => "B" // データソースを識別するためのタイプ。
      }
  }
  output {
      if[type] == "A" {
          elasticsearch {
              hosts => "http://es-cn-0h****dd0hcbnl.elasticsearch.aliyuncs.com:9200" // Elasticsearch クラスタのホスト。
              index => "rds_es_dxhtest_datetime_A" // Elasticsearch インデックス。
              user => "elastic" // Elasticsearch ユーザー名。
              password => "xxxxxxx" // Elasticsearch パスワード。
              document_id => "%{id}" // ドキュメント ID。
          }
      }
      if[type] == "B" {
          elasticsearch {            
              hosts => "http://es-cn-0h****dd0hcbnl.elasticsearch.aliyuncs.com:9200" // Elasticsearch クラスタのホスト。
              index => "rds_es_dxhtest_datetime_B" // Elasticsearch インデックス。
              user => "elastic" // Elasticsearch ユーザー名。
              password => "xxxxxxx" // Elasticsearch パスワード。
              document_id => "%{id}" // ドキュメント ID。
          }
      }
  }

  前の例では、type パラメーターが jdbc 設定に追加されています。 このパラメーターは、output 設定で条件を定義し、異なるテーブルのデータを異なるインデックスに同期するのに役立ちます。