背景信息

阿里云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網關實現公網數據傳輸。
  MySQL地址不正確，端口不是3306。	參見查看和管理實例連接地址和端口，獲取正確的地址和端口。使用正確的地址和端口，按照腳本格式替換管道配置中的jdbc_connection_string參數值。 重要 <MySQL的連接地址>：需要配置MySQL的內網地址。如果使用外網地址，需要為Logstash配置NAT網關實現公網數據傳輸，具體操作請參見配置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中進行判斷，將不同表的數據同步至不同的索引中。