elasticsearch-modelのimportメソッドのtransformパラメータは結構使える子

仕事で検索機能の改善をやってたときのメモを、秘匿事項を含まない部分だけ公開しとこうかなと思う。

Hoge.__elasticsearch__.import の定義

ポイントとしては

elasticsearch-model/lib/elasticsearch/model/importing.rb

          __find_in_batches(options) do |batch|
            response = client.bulk \
                         index:   target_index,
                         type:    target_type,
                         body:    __batch_to_bulk(batch, transform)

            yield response if block_given?

            errors +=  response['items'].select { |k, v| k.values.first['error'] }
          end

内部で、find_in_batchを使って1000件ずつ、Elastcsearchのbulk APIを使ってデータを一括投入するというスグレモノ。
（余談だが、find_in_batchesのデフォルトが1000なのは、Elasticsearch-modelではなくActiveRecord側の定義。 https://github.com/rails/rails/blob/6-0-stable/activerecord/lib/active_record/relation/batches.rb#L126 ）

bulk insertだけじゃなくてupdate, deleteもできる

Elasticsearchのbulk APIは、たとえば

[
  { delete: { id: 123} },
  { index: { id: 124, data: { title: elasticsearch-modelが便利, ......... } } },
  { delete: { id: 125 } }
]

のようなペイロードを指定すると、

• 123, 125はインデックスされていたら削除する(delete if exists)
• 124は、まだインデックスされていなかったらインデックスする（insert or update）

といった複雑なこともできる。

先に書いたimportメソッドの中で、bulk APIに渡されているペイロードは body: __batch_to_bulk(batch, transform) で、その定義を更に見ると

__batch_to_bulk

        def __batch_to_bulk(batch, transform)
          batch.map { |model| transform.call(model) }
        end

transformっていうのをmodelに適用したもののリストが渡されるようだ。

このtransformは、デフォルトだと
https://github.com/elastic/elasticsearch-rails/blob/v6.0.0/elasticsearch-model/lib/elasticsearch/model/adapters/active_record.rb#L114
に定義があるように

__transform

          def __transform
            lambda { |model|  { index: { _id: model.id, data: model.__elasticsearch__.as_indexed_json } } }
          end

モデルをそのままinsert or updateするような変換になっている。

たとえば、下書きと公開後でモデルを分けていないようなArticleみたいなモデルがあると、この変換は結構使えて、

公開済みと下書きを分けてimportする

# 公開済みの記事はインデックスする
Article.__elasticsearch__.import(scope: :published)

# 下書き状態のもの、公開後に下書きに戻されたものは、インデックスがある場合には削除する
Article.__elasticsearch__.import(
  scope: :draft,
  transform: ->(model) {
    { delete: { _id: model.id } }
  }
)

のようにtransform指定することで、公開済みのものだけを選択的にElasticsearch側に同期することができる。

あるいは、（可読性とのバランス次第だけど）

公開済みと下書きを分けてimportする②

Article.__elasticsearch__.import(
  transform: lambda do |article|
    if article.published?
      { index: { _id: article.id, data: article.__elasticsearch__.as_indexed_json } }
    else
      { delete: { _id: article.id } }
    end
  end
)

のようにすればbulk APIを叩く回数自体を減らすこともできそう。

　

そもそも、Webサービス(Rails)＋スマホアプリを作るときにやってはいけないこと 2018年版でも書いたんだけど、モデル分けようね！って話なんだけどさ... 歴史的な経緯とかもろもろでそうはいかないときには、transformパラメータはかなり使える子だと思う。