Elasticsearch と Rails のデータ同期方法

ども、@kimihom です。

全文検索の仕組みとして Elasticsearch を使ってサービスを運用している。Elasticsearch と Rails を使っている上で考慮しなきゃいけないデータ同期の方法について、それぞれのメリット/デメリットを紹介した上で最終的な提案まで記す。

本記事は Elasticsearch-Rails を利用した場合となるので、基本的な使用方法などはリンク先を参照いただければ幸いだ。

GitHub - elastic/elasticsearch-rails: Elasticsearch integrations for ActiveModel/Record and Ruby on Rails

では ActiveRecord と Elasticsearch でのデータ同期方法についていくつかご紹介しよう。

方法1. ActiveRecordコールバックによる同期

まず一番簡単な elasticsearch-model を使った ActiveRecord コールバックの方法だ。これは、ActiveRecord オブジェクトで作成/更新/削除されたタイミングで Elasticsearch のデータもそれぞれ更新する方法である。

class Article < ActiveRecord::Base
  include Elasticsearch::Model

  after_commit on: [:create] do
    __elasticsearch__.index_document
  end

  after_commit on: [:update] do
    __elasticsearch__.update_document
  end

  after_commit on: [:destroy] do
    __elasticsearch__.delete_document
  end
end

ほとんど登録/更新処理のないサービスの場合だったら、この方法で十分だろう。まずは ActiveRecord コールバックで問題ないか試しにやってみるといいと思う。

この方法のデメリットは2つある。1つは ActiveRecord オブジェクトの作成/更新のたびに Elasticsearch との接続が走ってしまって1つ1つのリクエストが重くなる点だ。当然 Rails サーバー側と Elasticsearch のサーバーは分けることになると思うので、その間のネットワーク接続で処理を待たされることになる。 2つめは ActiveRecord 限定で使っていかないといけない点だ。例えば裏側で PostgreSQL を使っていて、生の SQL でレコードをアップデートしても Elasticsearch 側に反映させることができない。あくまで #update や #create などの ActiveRecord のメソッドを呼ばない限り Elasticsearch に登録されない。これは運用時に結構ネックになることがあるので気をつけたいところだ。

方法2. バッチ処理で一括取り込みを行う

もう一つの方法は elasticsearch-rails を使って一括取り込みを行う方法だ。

これであれば、前者のパフォーマンス問題と、生SQL で同期できない問題を解決できる。具体的には、以下の rake タスクを定期的に実行するような形となる。

bundle exec rake environment elasticsearch:import:all

これは、include Elasticsearch::Model が読み込まれた ActiveRecord モデルを一括で読み込み、その全データを Elasticsearch に投入することを意味する。全データなので、これが1万件くらいあると、当然無駄に時間がかかってしまう。そして時間がかかるということで 1時間に1回の実行にしてしまうと、その間に登録されたデータは検索対象に入らなくなってしまう。この問題をなんとか解決したいところだ。

方法3. コールバックとバッチ処理の混合技

最終的なオススメは、上記2つを混合させる方法である。方法2で紹介した rake タスクは、対象の scope を指定することが可能だ。

bundle exec rake environment elasticsearch:import:all SCOPE='recent'

ActiveRecord 側で今回適当に用意した recent スコープを定義してあげる。

class Article < ActiveRecord::Base
  include Elasticsearch::Model

  scope :recent, -> {where("updated_at > ?", 1.hour.ago}

end

これで更新日時が 1時間以内のデータを一括で取ってきて、それだけを Elasticsearch に登録することが可能だ。しかし、この方法をしてしまうと、データ削除に対応できない という新たな問題が発生する。方法2の一括読み込みの方法では削除されてもデータ同期ができたが、今回は差分だけなので削除したレコードを取ってこれないのである。

てことで、削除の時だけ方法1のコールバックを活用しよう。今回の結論としては以下のような形となる。

class Article < ActiveRecord::Base
  include Elasticsearch::Model

  scope :recent, -> {where("updated_at > ?", 1.hour.ago}

  after_commit on: [:destroy] do
    __elasticsearch__.delete_document
  end
end

そして以下のコマンドを10分に1回などの Cron で回すようにしよう。

bundle exec rake environment elasticsearch:import:all SCOPE='recent'

1時間前の更新日時にしたのは、その間で Elasticsearch の同期に失敗しても、10分に1回なら6回のチャンスを与えるためにそうした。 import 自体は重複しても上書きするだけなので、確実にデータ同期する上でもこの方法で良いと思う。そしてこの方法なら、PostgreSQL の生 SQL でレコードをアップデートしたとしても、updated_at を正しく更新さえずれば Elasticsearch のデータ同期対象となる点も魅力的だ。 ※外部SQLでの削除 or ActiveRecord#delete(_all) 等は同期無理なのでやらないでね。