Ruby on Rails集成OpenSearch迁移方案咨询

1. 替换客户端依赖 #

放弃 elasticsearch-model 和 elasticsearch-rails，改用 OpenSearch 官方 Ruby 客户端。在 Gemfile 中添加：

gem 'opensearch-ruby'
gem 'opensearch-api'

执行 bundle install 完成依赖安装。

2. 重构模型中的索引逻辑 #

移除原有的 Elasticsearch 模块引入，手动实现索引配置、数据序列化、回调及批量导入逻辑，示例重构后的 Article 模型如下：

require 'opensearch'

class Article < ActiveRecord::Base
  # 初始化 OpenSearch 客户端
  def self.opensearch_client
    @opensearch_client ||= OpenSearch::Client.new(
      hosts: ['http://your-opensearch-host:9200'],
      user: 'your-username', # 若开启身份验证则添加
      password: 'your-password'
    )
  end

  # 定义索引设置与映射
  def self.index_config
    {
      settings: {
        number_of_shards: 1,
        max_result_window: MAX_RESULT_WINDOW,
        analysis: {
          filter: {
            email: {
              type: 'pattern_capture',
              preserve_original: 1,
              patterns: [
                '([^@]+)',
                '(\\p{L}+)',
                '(\\d+)',
                '@(.+)',
                '([^-@]+)'
              ]
            }
          },
          analyzer: {
            email: {
              tokenizer: 'uax_url_email',
              filter: ['email', 'lowercase', 'unique']
            }
          }
        }
      },
      mappings: {
        properties: {
          sent_at: {
            type: 'date',
            fields: { raw: { type: 'date', index: false } } # OpenSearch 用 index: false 替代 not_analyzed
          },
          id: { type: 'integer' },
          send_at: { type: 'date' },
          student_email: { type: 'text', analyzer: 'email' },
          subject: { type: 'text' },
          deleted_at: { type: 'date' }
        }
      }
    }
  end

  # 序列化待索引数据
  def as_indexed_json(options = {})
    self.as_json(
      only: [:id, :send_at],
      methods: [:student_email, :subject, :deleted_at]
    )
  end

  # 手动实现数据变更时的索引同步回调
  after_save :sync_to_opensearch
  after_destroy :remove_from_opensearch

  def sync_to_opensearch
    self.class.opensearch_client.index(
      index: 'articles',
      id: self.id,
      body: self.as_indexed_json
    )
  end

  def remove_from_opensearch
    self.class.opensearch_client.delete(
      index: 'articles',
      id: self.id
    )
  end

  # 创建索引方法
  def self.create_opensearch_index
    if opensearch_client.indices.exists?(index: 'articles')
      opensearch_client.indices.delete(index: 'articles')
    end
    opensearch_client.indices.create(
      index: 'articles',
      body: index_config
    )
  end

  # 批量导入数据，替代原 Article.import 功能
  def self.import_to_opensearch(batch_size = 1000)
    Article.find_in_batches(batch_size: batch_size) do |batch|
      bulk_ops = batch.flat_map do |article|
        [{ index: { _index: 'articles', _id: article.id } }, article.as_indexed_json]
      end
      opensearch_client.bulk(body: bulk_ops)
    end
  end
end

3. 替换原初始化逻辑 #

将原有的：

Article.__elasticsearch__.create_index!
Article.import

替换为：

Article.create_opensearch_index
Article.import_to_opensearch

4. 适配 OpenSearch 与 Elasticsearch 的差异 #

• 映射字段：OpenSearch 中 index: 'not_analyzed' 已被 index: false 或 type: 'keyword' 替代，需按实际需求调整。
• API 兼容：OpenSearch 兼容 Elasticsearch 7.x 大部分 API，但部分旧语法需微调，比如查询 DSL 的部分参数。
• 身份验证：若集群开启了账号密码验证，必须在客户端初始化时传入 user 和 password 参数。

5. 测试验证 #

• 在测试环境创建索引并导入数据，检查索引结构是否符合预期。
• 测试数据增删改操作，验证 OpenSearch 数据同步是否正常。
• 对比原 Elasticsearch 的搜索结果，确认搜索逻辑一致性。

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