Поиск записей mongoDB в пакетном режиме (с использованием адаптера mongoid ruby)

Question

Используя рельсы 3 и mongoDB с адаптером Mongoid, как я могу выполнить пакетное нахождение в DB Mongo?Мне нужно собрать все записи в конкретной коллекции БД mongo и проиндексировать их в solr (начальный индекс данных для поиска).

Проблема, с которой я сталкиваюсь, состоит в том, что, делая Model.all, собирает все записи исохраняет их в памяти.Затем, когда я обрабатываю их и индексирую в solr, моя память съедается, и процесс умирает.

Я пытаюсь сделать так, чтобы пакет находился в монго, чтобы я мог перебрать более 1000 записей за раз.время, передайте их solr в index, а затем обработайте следующую 1000 и т. д. *

Код, который у меня сейчас есть, делает это:

Model.all.each do |r|
  Sunspot.index(r)
end

Для коллекции, имеющей около 1,5миллион записей, это съедает 8+ ГБ памяти и убивает процесс.В ActiveRecord есть метод find_in_batches, который позволяет мне разбивать запросы на управляемые пакеты, что предотвращает выход памяти из-под контроля.Тем не менее, я не могу найти ничего подобного для mongoDB / mongoid.

Мне бы хотелось иметь возможность сделать что-то вроде этого:

Model.all.in_batches_of(1000) do |batch|
  Sunpot.index(batch)
end

Это бы облегчило мои проблемы с памятьюи справляться с трудностями, выполняя только управляемые задачи, установленные каждый раз.Однако документации по выполнению пакетных поисков в mongoDB немного.Я вижу много документации о выполнении пакетных вставок, но не о пакетных поисках.

Answer 1

В Mongoid вам не нужно вручную пакетировать запрос.

В Mongoid Model.all возвращает экземпляр Mongoid::Criteria.После вызова #each по этому критерию создается экземпляр драйвера Mongo, который используется для перебора записей.Этот базовый курсор драйвера Mongo уже объединяет все записи.По умолчанию batch_size равно 100.

Для получения дополнительной информации по этой теме прочитайте этот комментарий автора и сопровождающего Mongoid .

Таким образом, вы можете просто сделать это:

Model.all.each do |r|
  Sunspot.index(r)
end

Answer 2

Если вы выполняете итерацию по коллекции, где каждая запись требует большой обработки (т. Е. Запрашивает внешний API для каждого элемента), то курсор может истечь. В этом случае вам нужно выполнить несколько запросов, чтобы не оставлять курсор открытым.

require 'mongoid'

module Mongoid
  class Criteria
    def in_batches_of(count = 100)
      Enumerator.new do |y|
        total = 0

        loop do
          batch = 0

          self.limit(count).skip(total).each do |item|
            total += 1
            batch += 1
            y << item
          end

          break if batch == 0
        end
      end
    end
  end
end

Вот вспомогательный метод, который вы можете использовать для добавления функции пакетирования. Его можно использовать так:

Post.all.order_by(:id => 1).in_batches_of(7).each_with_index do |post, index|
  # call external slow API
end

Просто убедитесь, что у вас ВСЕГДА есть order_by по вашему запросу. В противном случае пейджинг может не выполнить то, что вы хотите. Также я бы придерживался партий по 100 или меньше. Как сказано в принятом ответе, запросы Mongoid в пакетах по 100, поэтому вы не хотите оставлять курсор открытым во время обработки.

Answer 3

Также быстрее отправлять партии в солнечное пятно. Вот как я это делаю:

records = []
Model.batch_size(1000).no_timeout.only(:your_text_field, :_id).all.each do |r|
  records << r
  if records.size > 1000
    Sunspot.index! records
    records.clear
  end
end
Sunspot.index! records

no_timeout: предотвращает отключение курсора (по умолчанию через 10 минут)

only: выбирает только идентификатор и поля, которые фактически проиндексированы

batch_size: получить 1000 записей вместо 100

Answer 4

Я не уверен насчет пакетной обработки, но вы можете сделать это так

current_page = 0
item_count = Model.count
while item_count > 0
  Model.all.skip(current_page * 1000).limit(1000).each do |item|
    Sunpot.index(item)
  end
  item_count-=1000
  current_page+=1
end

Но если вы ищете идеальное решение для длительного времени, я бы не рекомендовал это. Позвольте мне объяснить, как я справился с тем же сценарием в моем приложении. Вместо выполнения пакетных заданий

• я создал resque задание, которое обновляет индекс solr

  class SolrUpdator
   @queue = :solr_updator
  
   def self.perform(item_id)
     item = Model.find(item_id)
     #i have used RSolr, u can change the below code to handle sunspot
     solr = RSolr.connect :url => Rails.application.config.solr_path
     js = JSON.parse(item.to_json)
     solr.add js         
   end
  

  конец

• После добавления элемента я просто помещаю запись в очередь восстановления

  Resque.enqueue(SolrUpdator, item.id.to_s)
  

• Вот и все, запустите реск, и он позаботится обо всем

Answer 5

Следующее будет работать для вас, просто попробуйте

Model.all.in_groups_of(1000, false) do |r|
  Sunspot.index! r
end

Answer 6

Как сказал @RyanMcGeary, вам не нужно беспокоиться о пакетировании запроса. Однако индексирование объектов по одному намного медленнее, чем их пакетирование.

Model.all.to_a.in_groups_of(1000, false) do |records|
  Sunspot.index! records
end