How We Backfilled 2 Million Database Documents

1
Backfill.run(2000000, 5)

1
Backfill.run(100, 5)

1
def self.run(batch_size, thread_pool_size)
2
  jobs = Queue.new
3
  
4
  # get all the documents that meet these criteria
5
  objs = Obj.where(...)
6
  # limit the returned objects to the batch_size
7
  objs = objs.limit(batch_size)
8
  # push each document into the jobs queue to be processed
9
  objs.each { |o| jobs.push o }
10
  
11
  # create a thread pool
12
  workers = (thread_pool_size).times.map do |i|
13
    Thread.new do
14
      begin
15
        while j = jobs.pop(true)
16
          # log the thread index and object ID
17
          Rails.logger.with_fields(thread: i, obj: obj.id)
18
          begin
19
            # process objects
20
          end
21
...

1
thread="4" obj="939bpca..."
2
thread="1" obj="939apca..."
3
thread="5" obj="939cpca..."
4
thread="2" obj="939dpca..."
5
thread="3" obj="939fpca..."
6
thread="4" obj="969bpca..."
7
thread="1" obj="969apca..."
8
thread="5" obj="969cpca..."
9
thread="2" obj="969dpca..."
10
thread="3" obj="969fpca..."

1
thread="4" obj="989bpca..."
2
thread="1" obj="989apca..."
3
thread="4" obj="979bpca..."
4
thread="1" obj="979apca..."

1
          begin
2
            # process documents
3
          rescue
4
            next
5
          end

1
def self.run(batch_size, thread_pool_size, not_before)
2
  jobs = Queue.new
3
  
4
  # order the query results in ascending order
5
  objs = Obj.where(...).order(created_at: -1)
6
  # get documents created after the not_before date
7
  objs = objs.where(:created_at.gte => not_before)
8
  # limit the returned documents to the batch_size
9
  objs = objs.limit(batch_size)
10
  # push each document into the jobs queue to be processed
11
  objs.each { |o| jobs.push o }
12
13
  workers = (thread_pool_size).times.map do |i|
14
    Thread.new do
15
      begin
16
        while j = jobs.pop(true)
17
          # log each document's created_at date as it's processed
18
          Rails.logger.with_fields(thread: i, obj: obj.id, created_at: obj.created_at)
19
          begin
20
            # process documents
21
          rescue
22
            next
23
          end
24
...

1
  skip_value = batch_size
2
  step = batch_size
3
  
4
  loop do
5
    if jobs.size < 1000
6
      objs = Obj.where(...).order(created_at: -1)
7
      objs = objs.where(:created_at.gte => created_at)
8
      objs = objs.skip(skip_value).limit(step) # <--- job dies when this skip_value gets too big ❌
9
      objs.each { |r| jobs.push r }
10
      
11
      skip_value += step # <--- this keeps increasing as we process more objects ❌
12
  
13
      if objs.count == 0
14
        break
15
      end
16
    end
17
  end

1
  refill_at = 1000
2
  step = batch_size
3
  
4
  loop do
5
    if jobs.size < refill_at
6
      objs = Obj.where(...).order(created_at: -1)
7
      objs = objs.where(:created_at.gte => last_created_at) # <--- grab last_created_at constant from earlier in the script ✅
8
      objs = objs.skip(refill_at).limit(step) # <--- skip a constant amount ✅
9
      objs.each { |r| jobs.push r }
10
  
11
      if objs.count == 0
12
        break
13
      end
14
    end
15
  end

1
def self.run(batch_size, thread_pool_size, not_before)
2
  jobs = Queue.new
3
  
4
  objs = Obj.where(...).order(created_at: -1)
5
  objs = objs.where(:created_at.gte => not_before)
6
  objs = objs.limit(step)
7
  objs.each { |o| jobs.push o }
8
9
  updated = 0
10
  last_created_at = "" # <--- we update this variable... 
11
  
12
  workers = (thread_pool_size).times.map do |i|
13
    Thread.new do
14
      begin
15
        while j = jobs.pop(true)
16
          Rails.logger.with_fields(thread: i, obj: obj.id, created_at: obj.created_at)
17
          begin
18
            # process documents
19
            updated += 1
20
            last_created_at = obj.created_at # <--- ...with each document processed
21
          rescue
22
            next
23
          end
24
        end
25
      end
26
    end
27
  end
28
      
29
  loop do
30
    skip_value = batch_size
31
    step = 10000
32
  
33
    if jobs.size < 1000
34
      objs = Obj.where(...).order(created: -1)
35
      objs = objs.where(:created_at.gte => not_before)
36
      objs = objs.skip(skip_value).limit(step)
37
  
38
      objs.each { |r| jobs.push r }
39
      skip_value += step
40
  
41
      if objs.count == 0
42
        break
43
      end
44
    end
45
  end 
46
  workers.map(&:join)
47
end