巨大なCSVファイルをRailsで処理させる

# Rails内からRakeタスクを呼び出すために必要
require 'rake'

# タスクを読み込む必要があるため、この処理でタスクを読み込む
Rails.application.load_tasks

# Rakeタスクを実行する
Rake::Task['csv:import'].execute(:id => batch.id)

# こんな書き方はしないけど、イメージとしてこんな感じ。
redirect_to action: :csv

rails g task csv

# CSVのインポートを行うタスク
namespace :csv do
  require 'csv'

  # Railsから呼び出される特殊なタスク
  task :import => :environment do |task, args|
    Batch.find_by_id(args[:id])

    # CSV取り込み処理(最後に完全版を載せておきます。）
  end
end

Rake::Task['csv:import'].execute

system("cd #{Rails.root} & rake csv:import[#{batch.id}]")
redirect_to action: :csv

# CSVのインポートを行うタスク
namespace :csv do
  require 'csv'

  # Railsから呼び出される特殊なタスク
  # [:id]を付与して、第一引数に名前を付けた。
  task :import, [:id] => :environment do |task, args|
    Batch.find_by_id(args[:id])

    # CSV取り込み処理(最後に完全版を載せておきます。）
  end
end

`"cd #{Rails.root} & rake csv:import[#{batch.id}]"`
redirect_to action: :csv

fork do
    Process.setsid
    system("cd #{Rails.root} & rake csv:import[#{batch.id}]")
end

redirect_to action: :csv

# バルクインサートで使用
gem 'activerecord-import'
# バッチ一覧表示で使用
gem 'kaminari'
# 0.4系は落ちる・・・。
gem 'mysql2', '0.3.19'

$ rails generate model Batch

class CreateBatches < ActiveRecord::Migration
  def change
    create_table :batches do |t|
      # 処理しているバッチ名
      t.string :name, null: false
      # 処理ステータス
      t.integer :status, null: false, default: Batch::STATUS_READY
      # pid（ホスト名も入れると良いかもしれない。）
      t.string :pid
      # 画面に通知するメッセージ
      t.text :message
      # Rakeタスク内で絶対パスを取得するために使用
      t.text :rails_root_path

      t.timestamps null: false
    end
  end
end

# バッチの進捗状況を管理します。
class Batch < ActiveRecord::Base
  # バッチ名の定義
  # modelの部分にActiveRecodeのモデルクラスを定義する事で汎用的に使用できます。
  TYPE = {"csv_user" => {
                    name: "ユーザー取り込みバッチ",
                    model: User},
                "csv_category" => {
                    name: "カテゴリー取り込みバッチ",
                    model: Category}
  }

  #####
  # バッチのステータス
  #

  ## 開始前
  STATUS_READY = 0
  ## 処理中
  STATUS_PROCESSING_VALID = 1   # 検証中
  STATUS_PROCESSING_IMPORT = 2  # インポート中
  ## この値より大きい場合は処理終了とみなすための値
  STATUS_FINISH = 5
  ## 異常処理終了
  STATUS_FAIL = 8
  ## 正常処理終了
  STATUS_SUCCESS = 9

  # 処理終了時に保存している管理ファイルを削除する。
  after_save do
    # 処理が終了しているか判断
    if self.status > STATUS_FINISH
      # ファイルを削除する。
      begin
        File.unlink file_path
      rescue
        # 同一サーバーで処理をしているので、ファイルが無い事は無い。
        # ただし、念のため例外が発生しないようにしている。
      end
    end
  end

  # 「1/サーバー台数」の確率になるが、やらないよりマシなのでバッチがキャンセルされる時は指定のファイルを削除する。
  after_destroy do
    begin
      File.unlink file_path
    rescue
      # 別のサーバーにある場合はファイルを消せない。（通常運用をしている場合は管理しているファイルは削除されているはずなので、消せない場合はあまりないと割り切って諦める。）
    end
  end

  # CSVインポートで使用するモデルを取得します。
  def self.use_model(batch_type)
    # 指定のタスクが存在するかチェックします。
    raise "[csv_#{batch_type}] task is nothing!" if TYPE["csv_#{batch_type}"].nil?

    TYPE["csv_#{batch_type}"][:model]
  end

  # バッチの処理を行う。
  def self.csv_import_run(batch_type, file)
    # 形式チェックで使用。
    Batch.get_csv_model(batch_type)

    # バッチクラスを生成します。
    # Rake処理中にも絶対パスが取得できるようにRailsRootパスを保存しておく。（念のため）
    batch = Batch.new(name: "csv_#{batch_type}", rails_root_path: Rails.root, message: "CSVファイル読み込み中...")
    batch.save

    # バックグランドへ受け渡すためにファイルを保存します。
    batch.save_file file

    # Rakeタスクを別プロセスで実行させる。
    # ２度保存する処理が少し無駄だが、CSVアップロードをプロセス内で行うよりマシ。
    batch.pid = Batch.system_command("cd #{Rails.root} & rake csv:import[#{batch.id}]")
    batch.save
  end

  # バッチ名を取得
  def show_name
    NAME[self.name][:name]
  end

  # バッチのステータスを返却
  def show_status
    case self.status
      when STATUS_READY
        "取り込み準備中"
      when STATUS_PROCESSING_VALID
        "CSVデータ検証中"
      when STATUS_PROCESSING_IMPORT
        "CSV取り込み中"
      when STATUS_FAIL
        "取り込み処理失敗"
      when STATUS_SUCCESS
        "取り込み完了"
      else
        "不明な処理ステータス"
    end
  end

  # 保存するファイルパスを返却します。
  def save_file(file)
    # ファイルを書き込みます。
    File.open(self.file_path, 'wb') do|f|
      f.write(file.read)
    end
  end

  # このバッチが管理するファイルパスを返却
  # idで管理させるため、Batchレコードを作成後に呼び出す事
  def file_path
    # パスの定義
    path = "#{self.rails_root_path}/tmp/batch"

    # ファイルパスがない場合は作成する。
    FileUtils.mkdir_p(path) unless FileTest.exist?(path)

    # ファイルパスにIDを付けて返却
    "#{path}/#{self.id}"
  end

  # 処理が終了しているか？
  def finish?
    # 処理実行時間から1日以上経っている場合は処理終了とみなす。
    return true if self.updated_at < DateTime.now - 1.day
    # 終了ステータスの場合は処理終了とみなす。
    self.status > STATUS_FINISH
  end

  private

  # システムコマンドを別プロセスで実施します。
  def self.system_command(command)
    # 念のためログを出力
    Rails.logger.info command

    # 別プロセスで実行させる。
    pid = nil
    fork do
      Process.setsid
      fork do
        # PIDの取得
        pid = `#{command} 2>&1 & echo $!`.lstrip.chomp
        if pid =~ /[0-9]/
          Rails.logger.info "csv import rake task pid = #{pid}"
        else
          Rails.logger.error 'command has not pid'
        end
      end
    end
    # PIDの返却（僕はpidをログに出力までしかしてません。）
    pid
  end
end

"account","name","memo"
"test1","テスト太郎","このメモは取り込みたくない。"
...

# ユーザーデータ
class User < ActiveRecord::Base
  # validate
  validates :account, :name, presence: true

  # 使用カラムを取得します。
  def self.use_columns
    # シンボルは不可（memoは取り込み対象から外す。）
    ["account", "name"]
  end

  # PKカラムを返却します。(同じデータがある場合は上書きをするため)
  def self.pk_columns
    # シンボル可
    [:account]
  end

  # CSVデータの検証を行います。
  def csv_valid?(params)
    begin
      self.attributes = params
    rescue => e
      logger.error("user import error:#{e}")
    end
    self.valid?
  end
end

<%= form_tag csv_user_path, multipart: true do %>
  <div class="form-group">
    <label for="input_file">ユーザーデータ</label>
    <%= file_field_tag :file %>
    <p class="help-block">ex) user.csv</p>
  </div>
  <button type="submit" class="btn btn-default">アップロード</button>
<% end %>

$ rails generate controller Csv

class CsvController < ApplicationController
  # 実際は汎用的に使っているため、幾つか抽象化するための変数があります。
  def user_csv_upload
    # 汎用的に呼び出されるため事前に。
    file = params[:file]
    batch_type = :user
    return flash[:error] = "ファイルを選択してください。" unless file

    # 一行目のデータをチェックする場合は以下の処理を行います。
    # バッチ処理でも同じ事をするので冗長の場合はmodule化して使ってください。
    begin
      # モデルを取得します。
      model = Batch.use_model(batch_type)

      open(file.path, 'r:utf-8:utf-8', undef: :replace) do |f|
        csv = ::CSV.new(f, :headers => :first_row)
        csv.each do |row|
          next if row.header_row?

          # CSVの行情報をHASHに変換
          table = Hash[[row.headers, row.fields].transpose]

          # 指定パラメータのデータのみを対象とします。
          params = table.to_hash.slice(*model.use_columns)

          # 対象データを取得します。（ここでパラメータを受け渡してもOK）
          instance = model.new

          # データの整合性チェックをします。
          # ※各モデル毎にvalid?の仕組みが違う場合はcsv_valid?を使用する事をお勧めします。
          unless instance.csv_valid?(params)
            # 失敗した場合は処理を抜けます。
            return flash[:error] = "ファイル形式が違います。#{instance.errors.full_messages}"
          end

          # ２行目以降のデータは取り込まない。
          break
        end
      end
    rescue => e
      return flash[:error] = "CSV取り込み処理でエラーが発生しました。#{e.message}"
    end

    # 処理をバッチに委ねます。
    Batch.csv_import_run(batch_type, file)
    flash[:info] = "CSVの取り込みを開始します。進捗状態は処理一覧で確認できます。"

    # 適当なURLにリダイレクトします。
    redirect_to action: :user
  end
end

$ rails g task csv

# CSVのインポートを行うタスク
namespace :csv do
  # CSVの処理を行うので`require`しておく。(Rails内部で行う場合は不要だが、Rakeで行う場合は必要)
  require 'csv'

  # Railsから呼び出される。
  task :import, [:id] => :environment do |task, args|
    # 読み込みデータ数
    total_count = 0

    # ファイルを取り込み開始します。
    batch = Batch.find_by_id(args[:id])
    batch.status = Batch::STATUS_PROCESSING_VALID
    batch.message = "#{total_count}件〜#{total_count+100}件目のチェックを行ってます。"
    batch_status.save

    # 使用するモデルを取得します。（設計ミス。use_modelが使えない。）
    model = Batch::TYPE[batch.name][:model]

    # CSVデータを読み込みます。
    begin
      open(batch.file_path, 'r:utf-8:utf-8', undef: :replace) do |f|
        data_list = []
        csv = CSV.new(f, :headers => :first_row)
        csv.each do |row|
          next if row.header_row?

          # CSVの行情報をHASHに変換
          table = Hash[[row.headers, row.fields].transpose]

          # パラメータを取得します。
          params = table.to_hash.slice(*model.use_columns)

          # 対象データを取得します。
          instance = model.new

          # データの検証を行います。
          unless instance.csv_valid?(params)
            # 失敗した場合は処理を抜けます。
            batch.message = "#{total_count+1}件目でエラーが発生しました。#{instance.errors.full_messages}"
            batch.status = Batch::STATUS_FAIL
            batch.save
            return
          end

          # 保存するデータを追加します。
          data_list << instance

          # 取り込み件数を増やします。
          total_count += 1

          # チェック件数が10000件を超えるたびにメッセージを更新する。
          if total_count % 10000 == 0
            batch.status = Batch::STATUS_PROCESSING_IMPORT
            batch.message = "CSVの取り込みを行ってます。"
            batch.save
            model.import data_list, on_duplicate_key_update: model.pk_columns, validate: false
            data_list = []
            batch.status = Batch::STATUS_PROCESSING_VALID
            batch.message = "#{total_count}件〜#{total_count+100}件目のチェックを行ってます。"
            batch.save
            # 処理を占有しないように1秒まつ。（この方法で良いのだろうか・・・）
            sleep(1)
          end
        end

        # DB登録処理を行う。
        batch.status = Batch::STATUS_PROCESSING_IMPORT
        batch.message = "CSVの取り込みを行ってます。"
        batch.save
        model.import data_list, on_duplicate_key_update: model.pk_columns, validate: false
      end

      # CSV取り込み終了
      batch.status = Batch::STATUS_SUCCESS
      batch.message = "取り込み件数[#{total_count}]を取り込み完了しました。"
      batch.save
    rescue => e
      batch.status = Batch::STATUS_FAIL
      batch.message = "CSV取り込み処理でエラーが発生しました。#{e.message}"
      batch.save
    end
  end
end

@batches = Batch.page(params[:page]).order(status: :asc, updated_at: :desc)

<%= paginate(@batches) %>
<table class="table">
  <thead>
  <tr>
    <th>バッチ種別</th>
    <th>ステータス</th>
    <th>状況</th>
    <th>取り込み開始日</th>
    <th>最終処理更新日</th>
    <th>操作</th>
  </tr>
  </thead>
  <tbody>
  <% @batches.each { |batch| %>
      <tr>
        <td><%= batch.show_name %></td>
        <td><%= batch.show_status %></td>
        <td><%= batch.message %></td>
        <td><%= batch.created_at.strftime("%Y/%m/%d %H:%M:%S") %></td>
        <td><%= batch.updated_at.strftime("%Y/%m/%d %H:%M:%S") %></td>
        <td><%= batch.finish? ? link_to("削除", admin_csv_batch_destroy_path(batch_status.id), method: :delete) : "" %></td>
      </tr>
  <% } %>
  </tbody>
</table>