スクレイピングで集めたデータを外部APIで加工するバッチの設計：キャッシュ・レート制御・途中再開

「集めてから加工する」で従量コストと実行時間が爆発する
同じ入力を二度APIに投げないキャッシュ層を挟む
レート制限に合わせたsleepと429時のバックオフ
中間結果を逐次保存し、途中から再開できるようにする
「収集」と「加工」を別フェーズに分けると失敗の切り分けが楽になる
まとめ

「集めてから加工する」で従量コストと実行時間が爆発する

Seleniumで対象サイトを巡回してデータを集めたあと、そのままでは使いにくい生データを外部の従量課金API（翻訳や地理情報の変換のようなものを想像してください）に投げて整形・加工する、という2段構えのバッチをよく書きます。最初のうちは件数が少ないので気にならないのですが、収集件数が数千件を超えてくると、素朴に「1件ずつAPIを呼ぶ」実装は簡単に破綻します。

まず費用です。1件あたり数円のAPIでも、数千件×毎日実行となれば月間コストは無視できない額になります。次に実行時間です。API側のレスポンスに数百ミリ秒かかるとして、直列に1件ずつ呼んでいるだけで数千件の処理に数時間かかることも珍しくありません。さらに厄介なのは、この処理を実行するたびに「前回と同じ入力」に対してもう一度課金してAPIを叩き直してしまうことです。収集フェーズは差分更新でも、加工フェーズは毎回全件をやり直す実装にしてしまうと、コストと時間の両方が収集件数の増加にそのまま比例して膨らんでいきます。

同じ入力を二度APIに投げないキャッシュ層を挟む

対策の基本は、一度APIに投げた入力とその結果をローカルに保存しておき、同じ入力が来たらAPIを呼ばずにキャッシュを返すことです。メモリ上のHashだけでは実行のたびに消えてしまうため、JSONファイルに永続化して、次回起動時にも再利用できるようにします。

require 'json'
require 'net/http'
require 'uri'

class CachedApiClient
  def initialize(cache_path, endpoint: 'https://api.example.com/v1/convert')
    @cache_path = cache_path
    @endpoint = endpoint
    @cache = load_cache
  end

  def convert(input_text)
    key = input_text.to_s.strip
    return @cache[key] if @cache.key?(key)

    result = request_with_backoff(key)
    @cache[key] = result
    persist_cache
    result
  end

  private

  def load_cache
    return {} unless File.exist?(@cache_path)
    JSON.parse(File.read(@cache_path))
  rescue JSON::ParserError
    warn 'キャッシュファイルが壊れていたため空で再開します'
    {}
  end

  def persist_cache
    tmp_path = "#{@cache_path}.tmp"
    File.write(tmp_path, JSON.generate(@cache))
    File.rename(tmp_path, @cache_path)
  end
end

convertの入口でキャッシュを確認し、ヒットすればAPIを呼ばずに即座に返すだけで、2回目以降の実行では「新規に増えた分」だけが課金対象になります。キャッシュの保存も、書き込み中にプロセスが落ちてキャッシュファイル自体が壊れると本末転倒なので、一時ファイルに書いてからrenameする形にしています。この手法は以前、SeleniumバッチのCSVログをtmp→renameで壊れないように書くでCSVログを対象に紹介したものと同じ考え方です。

レート制限に合わせたsleepと429時のバックオフ

キャッシュで呼び出し回数を減らせても、新規分はまとまった件数を短時間に投げがちで、APIのレート制限に引っかかりやすくなります。通常時は決められた間隔でsleepを挟み、429 Too Many Requestsが返ってきたときは指数的に待ち時間を伸ばしながらリトライする、という2段構えの制御を入れます。

def request_with_backoff(key, attempt = 0)
  uri = URI(@endpoint)
  request = Net::HTTP::Post.new(uri)
  request['Content-Type'] = 'application/json'
  request.body = JSON.generate(text: key)

  response = Net::HTTP.start(uri.host, uri.port, use_ssl: true) { |http| http.request(request) }

  case response.code.to_i
  when 200
    JSON.parse(response.body)
  when 429
    raise "レート制限が解消しません（#{attempt}回目）" if attempt >= 5
    wait_seconds = (response['Retry-After'] || (2**attempt)).to_i
    sleep wait_seconds
    request_with_backoff(key, attempt + 1)
  else
    raise "APIエラー: #{response.code} #{response.body}"
  end
ensure
  sleep 0.3 # 通常時のレート制御。API側のドキュメントに合わせて調整する
end

Retry-Afterヘッダーがあればそれを優先し、なければ2**attemptで待ち時間を伸ばしていきます。リトライ回数に上限を設けているのは、レート制限が長時間解消しない異常事態を無限リトライで隠さず、上位の呼び出し側にきちんと例外として上げるためです。

中間結果を逐次保存し、途中から再開できるようにする

数千件をまとめて処理していると、ネットワーク断やAPI側の障害で途中で止まることが前提になります。ここで全件を最後にまとめて書き出す設計だと、99%処理し終えた時点で落ちても成果がゼロになってしまうため、1件処理するたびに結果をJSON Lines形式で追記し、再実行時にはすでに処理済みのIDをスキップするようにします。

def run_batch(scraped_items, output_path, cache_path)
  client = CachedApiClient.new(cache_path)
  done_ids = load_progress(output_path)

  File.open(output_path, 'a') do |file|
    scraped_items.each do |item|
      next if done_ids.include?(item[:id])

      begin
        converted = client.convert(item[:raw_text])
        file.puts(JSON.generate(id: item[:id], converted: converted))
        file.flush
      rescue => e
        warn "id=#{item[:id]} の加工でエラー: #{e.message}"
      end
    end
  end
end

def load_progress(output_path)
  return [] unless File.exist?(output_path)

  File.readlines(output_path).filter_map do |line|
    JSON.parse(line)['id']
  rescue JSON::ParserError
    nil
  end
end

再実行時はload_progressで処理済みIDの集合を作り直すだけで、途中から続きを処理できます。1行1レコードのJSON Linesにしているのは、末尾に追記するだけで良く、ファイル全体を読み直して書き直す必要がないためです。

「収集」と「加工」を別フェーズに分けると失敗の切り分けが楽になる

キャッシュ・レート制御・再開処理を入れても、Seleniumでの収集処理と外部APIでの加工処理を1つのループの中で同時にやってしまうと、失敗したときにどちらが原因か切り分けにくくなります。収集は収集で完結させてファイルに書き出し、加工は別のプロセスとしてそのファイルを読んで実行する形に分けておくと、「今日はスクレイピングでブロックされた」のか「今日はAPI側のレート制限に引っかかった」のかがログの出所ではっきり区別できます。フェーズを分けることで、加工フェーズだけを何度でも安全に再実行できるようになるのも大きな利点です。