通过Logstash将RDS MySQL数据同步至Elasticsearch

背景信息

阿里云Logstash是一款强大的数据收集和处理工具，提供了数据采集、转换、优化和输出的能力。通过Logstash的logstash-input-jdbc插件（默认已安装，不可卸载），可批量查询RDS MySQL中的数据并同步到Elasticsearch中。同时，logstash-input-jdbc插件会定期对RDS中的数据进行轮询查询，并将自上次轮询以来插入或更改的记录同步到Elasticsearch。更多详细信息，请参见官方文档中的如何使用Logstash和JDBC确保Elasticsearch与关系型数据库保持同步。本方案适用于同步全量数据且接受秒级延迟的场景或批量查询特定条件的数据然后进行同步的场景。

前提条件

已创建RDS MySQL实例、阿里云Elasticsearch实例、Logstash实例。建议您在同一专有网络下创建相关实例。

• 创建RDS MySQL实例。具体操作请参见创建RDS MySQL实例。本文以MySQL 5.7版本为例。

• 创建阿里云Elasticsearch实例。具体操作请参见创建阿里云Elasticsearch实例。本文创建的实例的版本为通用商业版6.7。

• 创建阿里云Logstash实例。具体操作参见创建阿里云Logstash实例。

  说明

  您也可以使用公网环境的服务，前提是需要配置SNAT、打开RDS MySQL的公网地址并取消白名单限制。SNAT的具体配置方法，请参见配置NAT公网数据传输。设置白名单操作，请参见设置IP白名单。

使用限制

操作步骤

常见问题

• 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中进行判断，将不同表的数据同步至不同的索引中。