如何实现AWS RDS与OpenSearch（Elasticsearch）的数据同步？

完全可以用Logstash实现RDS与OpenSearch的数据同步——OpenSearch作为Elasticsearch的分支，API层面与Elasticsearch高度兼容，因此你在本地Elasticsearch上的Logstash同步逻辑可以直接适配到OpenSearch。

核心配置逻辑 #

Logstash通过Input（读取RDS数据）、Filter（可选数据处理）、**Output（写入OpenSearch）**三个阶段完成同步，以下是具体配置细节：

1. Input阶段：JDBC插件读取RDS #

使用Logstash的jdbc输入插件连接RDS，支持全量/增量同步，需提前下载对应数据库的JDBC驱动（如MySQL的mysql-connector-java），配置RDS连接信息与同步策略（推荐增量同步，避免重复数据）。

2. Filter阶段：可选数据处理 #

可通过mutate等插件对数据进行字段重命名、类型转换、冗余字段移除等操作，按需配置。

3. Output阶段：写入OpenSearch #

利用Logstash的elasticsearch输出插件（兼容OpenSearch），配置OpenSearch端点与认证信息（AWS环境推荐IAM认证）。

完整配置示例 #

input {
  jdbc {
    # RDS连接字符串，替换为你的RDS端点、数据库名
    jdbc_connection_string => "jdbc:mysql://your-rds-endpoint:3306/your-db?useSSL=true&requireSSL=true"
    jdbc_user => "your-db-username"
    jdbc_password => "your-db-password"
    # JDBC驱动路径，替换为实际路径
    jdbc_driver_library => "/opt/logstash/drivers/mysql-connector-java-8.0.30.jar"
    jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
    
    # 增量同步配置：基于updated_at字段跟踪最后同步时间
    tracking_column => "updated_at"
    tracking_column_type => "timestamp"
    use_column_value => true
    last_run_metadata_path => "/opt/logstash/data/last_run.txt"
    
    # 仅同步上次同步后更新的数据
    statement => "SELECT * FROM your_table WHERE updated_at > :sql_last_value"
    # 同步频率：每分钟执行一次（cron表达式）
    schedule => "* * * * *"
  }
}

filter {
  mutate {
    # 将数据库id作为OpenSearch文档ID，避免重复写入
    rename => { "id" => "[@metadata][_id]" }
    # 移除Logstash默认生成的冗余字段
    remove_field => ["@version", "@timestamp"]
  }
}

output {
  elasticsearch {
    # 替换为你的OpenSearch端点
    hosts => ["https://your-opensearch-endpoint:443"]
    # 目标索引名称
    index => "your-target-index"
    document_id => "%{[@metadata][_id]}"
    
    # AWS OpenSearch认证配置
    aws_signature => true
    # 替换为OpenSearch所在区域
    region => "us-east-1"
    
    # 若未使用IAM角色，可手动配置密钥（不推荐硬编码）
    # access_key_id => "your-aws-access-key"
    # secret_access_key => "your-aws-secret-key"
  }
}

关键注意事项 #

• 驱动兼容性：JDBC驱动版本需与RDS数据库版本匹配（如MySQL 8.x需使用8.x系列驱动）
• 增量同步策略：优先使用时间戳（如updated_at）或自增ID作为跟踪字段，减少全量同步的性能开销
• AWS权限配置：若Logstash运行在EC2上，可为实例附加包含OpenSearch写入权限的IAM角色（如AmazonOpenSearchServiceFullAccess，或更细粒度的es:ESHttpPut等权限），无需硬编码密钥
• 网络连通性：确保Logstash所在环境（EC2/ECS等）能访问RDS与OpenSearch，可通过VPC peering、安全组规则配置实现
• 版本兼容：OpenSearch 1.x及以上版本与Elasticsearch 7.x API高度兼容，Logstash 7.x及以上版本均可适配

内容的提问来源于stack exchange，提问作者lakhr034