Elasticsearch：通過Logstash將RDS MySQL資料同步至Elasticsearch

背景資訊

阿里雲Logstash是一款強大的資料收集和處理工具，提供了資料擷取、轉換、最佳化和輸出的能力。通過Logstash的logstash-input-jdbc外掛程式（預設已安裝，不可卸載），可批量查詢RDS MySQL中的資料並同步到Elasticsearch中。同時，logstash-input-jdbc外掛程式會定期對RDS中的資料進行輪詢查詢，並將自上次輪詢以來插入或更改的記錄同步到Elasticsearch。更多詳細資料，請參見官方文檔中的如何使用Logstash和JDBC確保Elasticsearch與關係型資料庫保持同步。本方案適用於同步全量資料且接受秒級延遲的情境或批量查詢特定條件的資料然後進行同步的情境。

前提條件

建議您在同一專用網路下建立以下執行個體：

• 建立RDS MySQL執行個體。具體操作請參見建立RDS MySQL執行個體。本文以MySQL 5.7版本為例。

• 建立Elasticsearch執行個體。具體操作請參見建立Elasticsearch執行個體。本文以7.10版本ES執行個體為例。

• 建立阿里雲Logstash執行個體。具體操作參見建立阿里雲Logstash執行個體。

使用限制

操作步驟

常見問題

• Q：同步任務失敗（例如管道一直在生效中、前後資料不一致、資料庫連接不成功），如何解決？

  A：查看Logstash執行個體的主日誌是否有報錯，根據報錯判斷原因，具體操作請參見查詢日誌。常見的原因及解決方案如下。

  說明

  執行以下操作時，如果叢集正在變更中，可參見查看執行個體任務進度詳情先中斷變更，操作完成後再觸發重啟恢複。

  原因	解決方案
  MySQL白名單中沒有加入Logstash的IP地址。	參見通過用戶端、命令列串連RDS MySQL執行個體，在MySQL白名單中加入Logstash節點的IP地址。 說明 擷取Logstash的IP地址的具體操作，請參見查看執行個體的基本資料。
  Logstash的IP地址沒有添加到對應ECS伺服器的安全性群組中（ECS自建MySQL）。	參見添加安全性群組規則，在ECS安全性群組中添加Logstash的IP地址和連接埠號碼。 說明 擷取Logstash的IP地址和連接埠號碼的具體操作，請參見查看執行個體的基本資料。
  Logstash和Elasticsearch不在同一VPC下。	選擇以下任意一種方式處理： 參見建立Elasticsearch執行個體，重新購買同一VPC下的Elasticsearch執行個體。購買後，修改現有管道配置。 參見配置NAT公網資料轉送，配置NAT Gateway實現公網資料轉送。
  MySQL地址不正確，連接埠不是3306。	參見查看和管理執行個體串連地址和連接埠，擷取正確的地址和連接埠。使用正確的地址和連接埠，按照指令碼格式替換管道配置中的jdbc_connection_string參數值。 重要 <MySQL的串連地址>：需要配置MySQL的內網地址。如果使用外網地址，需要為Logstash配置NAT Gateway實現公網資料轉送，具體操作請參見配置NAT公網資料轉送。
  Elasticsearch未開啟自動建立索引。	參見配置YML參數，開啟Elasticsearch執行個體的自動建立索引功能。
  Elasticsearch或Logstash的負載太高。	參見升配叢集，升級執行個體規格。 說明 Elasticsearch負載情況可參見指標含義與異常處理建議，通過控制台監控指標查看。Logstash負載情況可參見配置X-Pack監控，通過Kibana X-Pack監控查看。
  沒有上傳JDBC串連MySQL的驅動檔案。	參見配置擴充檔案，下載並上傳驅動檔案。
  管道配置中包含了file_extend，但沒有安裝logstash-output-file_extend外掛程式。	選擇以下任意一種方式處理： 參見安裝或卸載外掛程式，安裝logstash-output-file_extend外掛程式。 在管道配置中，去掉file_extend配置。

  更多問題原因及解決方案，請參見Logstash資料寫入問題排查方案。

• Q：管道input配置中，如何在一個管道中配置多個源端JDBC？

  A：您可以在管道input配置中定義多個jdbc資料來源，並在statement中指定對應表的查詢語句，實現一個管道中配置多個源端JDBC，參考樣本如下。

  input {
      jdbc {
        jdbc_driver_class => "com.mysql.jdbc.Driver"
        jdbc_driver_library => "/ssd/1/share/<Logstash執行個體ID>/logstash/current/config/custom/mysql-connector-java-5.1.48.jar"
        jdbc_connection_string => "jdbc:mysql://rm-bp1xxxxx.mysql.rds.aliyuncs.com:3306/<資料庫名稱>?useUnicode=true&characterEncoding=utf-8&useSSL=false&allowLoadLocalInfile=false&autoDeserialize=false"
        jdbc_user => "xxxxx"
        jdbc_password => "xxxx"
        jdbc_paging_enabled => "true"
        jdbc_page_size => "50000"
        statement => "select * from tableA where update_time >= :sql_last_value"
        schedule => "* * * * *"
        record_last_run => true
        last_run_metadata_path => "/ssd/1/<Logstash執行個體ID>/logstash/data/last_run_metadata_update_time.txt"
        clean_run => false
        tracking_column_type => "timestamp"
        use_column_value => true
        tracking_column => "update_time"
        type => "A"
      }
      jdbc {
        jdbc_driver_class => "com.mysql.jdbc.Driver"
        jdbc_driver_library => "/ssd/1/share/<Logstash執行個體ID>/logstash/current/config/custom/mysql-connector-java-5.1.48.jar"
        jdbc_connection_string => "jdbc:mysql://rm-bp1xxxxx.mysql.rds.aliyuncs.com:3306/<資料庫名稱>?useUnicode=true&characterEncoding=utf-8&useSSL=false&allowLoadLocalInfile=false&autoDeserialize=false"
        jdbc_user => "xxxxx"
        jdbc_password => "xxxx"
        jdbc_paging_enabled => "true"
        jdbc_page_size => "50000"
        statement => "select * from tableB where update_time >= :sql_last_value"
        schedule => "* * * * *"
        record_last_run => true
        last_run_metadata_path => "/ssd/1/<Logstash執行個體ID>/logstash/data/last_run_metadata_update_time.txt"
        clean_run => false
        tracking_column_type => "timestamp"
        use_column_value => true
        tracking_column => "update_time"
        type => "B"
      }
  }
  output {
      if[type] == "A" {
          elasticsearch {
              hosts => "http://es-cn-0h****dd0hcbnl.elasticsearch.aliyuncs.com:9200"
              index => "rds_es_dxhtest_datetime_A"
              user => "elastic"
              password => "xxxxxxx"
              document_id => "%{id}"
          }
      }
      if[type] == "B" {
          elasticsearch {            
              hosts => "http://es-cn-0h****dd0hcbnl.elasticsearch.aliyuncs.com:9200"
              index => "rds_es_dxhtest_datetime_B"
              user => "elastic"
              password => "xxxxxxx"
              document_id => "%{id}"
          }
      }
  }

  以上樣本在jdbc中，新增了一個屬性type，用來在output中進行判斷，將不同表的資料同步至不同的索引中。