Rails:对许多记录执行更新的更快方法

Question

在我们的Rails 3.2.13应用程序(Ruby2.0.0+Heroku上的Postgres )中，我们经常从API中提取大量订单数据，然后需要更新或创建数据库中的每个订单以及关联。单个订单会创建/更新自身以及大约。10-15个复杂的对象，我们一次进口多达500个订单。

下面的代码可以工作，但问题是它在速度方面根本没有效率。创建/更新500条记录需要大约。1分钟，并生成6500+ db查询！

def add_details(shop, shopify_orders)
  shopify_orders.each do |shopify_order|
    order = Order.where(:order_id => shopify_order.id.to_s, :shop_id => shop.id).first_or_create
    order.update_details(order,shopify_order,shop)  #This calls update_attributes for the Order
    ShippingLine.add_details(order, shopify_order.shipping_lines)
    LineItem.add_details(order, shopify_order.line_items)
    Taxline.add_details(order, shopify_order.tax_lines)
    Fulfillment.add_details(order, shopify_order.fulfillments)
    Note.add_details(order, shopify_order.note_attributes)
    Discount.add_details(order, shopify_order.discount_codes)
    billing_address = shopify_order.billing_address rescue nil
    if !billing_address.blank?
      BillingAddress.add_details(order, billing_address)
    end
    shipping_address = shopify_order.shipping_address rescue nil
    if !shipping_address.blank?
      ShippingAddress.add_details(order, shipping_address)
    end
    payment_details = shopify_order.payment_details rescue nil
    if !payment_details.blank?
      PaymentDetail.add_details(order, payment_details)
    end
  end
end

  def update_details(order,shopify_order,shop)
    order.update_attributes(
      :order_name => shopify_order.name,
      :order_created_at => shopify_order.created_at,
      :order_updated_at => shopify_order.updated_at,
      :status => Order.get_status(shopify_order),
      :payment_status => shopify_order.financial_status,
      :fulfillment_status => Order.get_fulfillment_status(shopify_order),
      :payment_method => shopify_order.processing_method,
      :gateway => shopify_order.gateway,
      :currency => shopify_order.currency,
      :subtotal_price => shopify_order.subtotal_price,
      :subtotal_tax => shopify_order.total_tax,
      :total_discounts => shopify_order.total_discounts,
      :total_line_items_price => shopify_order.total_line_items_price,
      :total_price => shopify_order.total_price,
      :total_tax => shopify_order.total_tax,
      :total_weight => shopify_order.total_weight,
      :taxes_included => shopify_order.taxes_included,
      :shop_id => shop.id,
      :email => shopify_order.email,
      :order_note => shopify_order.note
    )
  end

因此，正如您所看到的，我们正在遍历每个订单，找出它是否存在(然后要么加载现有订单，要么创建新订单)，然后调用update_attributes传递订单的详细信息。之后，我们创建或更新每个关联。每个相关联的模型看起来都非常相似：

  class << self
    def add_details(order, tax_lines)
      tax_lines.each do |shopify_tax_line|
        taxline = Taxline.find_or_create_by_order_id(:order_id => order.id)
        taxline.update_details(shopify_tax_line)
      end
    end
  end
  def update_details(tax_line)
    self.update_attributes(:price => tax_line.price, :rate => tax_line.rate, :title => tax_line.title)
  end

我已经研究过activerecord，但不幸的是，它似乎更适合批量创建记录，而不是按照我们的要求进行更新。

提高性能的最佳方法是什么？

许多人事先表示感谢。

更新：

我提出了这个小小的改进，它本质上删除了更新新创建的订单的调用(每个订单减少一个查询)。

 def add_details(shop, shopify_orders)
      shopify_orders.each do |shopify_order|
      values = {:order_id => shopify_order.id.to_s, :shop_id => shop.id,
        :order_name => shopify_order.name,
            :order_created_at => shopify_order.created_at,
            :order_updated_at => shopify_order.updated_at,
            :status => Order.get_status(shopify_order),
            :payment_status => shopify_order.financial_status,
            :fulfillment_status => Order.get_fulfillment_status(shopify_order),
            :payment_method => shopify_order.processing_method,
            :gateway => shopify_order.gateway,
            :currency => shopify_order.currency,
            :subtotal_price => shopify_order.subtotal_price,
            :subtotal_tax => shopify_order.total_tax,
            :total_discounts => shopify_order.total_discounts,
            :total_line_items_price => shopify_order.total_line_items_price,
            :total_price => shopify_order.total_price,
            :total_tax => shopify_order.total_tax,
            :total_weight => shopify_order.total_weight,
            :taxes_included => shopify_order.taxes_included,
            :email => shopify_order.email,
            :order_note => shopify_order.note}
        get_order = Order.where(:order_id => shopify_order.id.to_s, :shop_id => shop.id)
        if get_order.blank?
            order = Order.create(values)
        else
        order = get_order.first  
            order.update_attributes(values)
        end
        ShippingLine.add_details(order, shopify_order.shipping_lines)
        LineItem.add_details(order, shopify_order.line_items)
        Taxline.add_details(order, shopify_order.tax_lines)
        Fulfillment.add_details(order, shopify_order.fulfillments)
        Note.add_details(order, shopify_order.note_attributes)
        Discount.add_details(order, shopify_order.discount_codes)
        billing_address = shopify_order.billing_address rescue nil
        if !billing_address.blank?
          BillingAddress.add_details(order, billing_address)
        end
        shipping_address = shopify_order.shipping_address rescue nil
        if !shipping_address.blank?
          ShippingAddress.add_details(order, shipping_address)
        end
        payment_details = shopify_order.payment_details rescue nil
        if !payment_details.blank?
          PaymentDetail.add_details(order, payment_details)
        end
      end
 end

对于相关的对象：

  class << self
    def add_details(order, tax_lines)
      tax_lines.each do |shopify_tax_line|
        values = {:order_id => order.id,
            :price => tax_line.price,
            :rate => tax_line.rate,
            :title => tax_line.title}
        get_taxline = Taxline.where(:order_id => order.id)
        if get_taxline.blank?
            taxline = Taxline.create(values)
        else
            taxline = get_taxline.first  
            taxline.update_attributes(values)
        end
      end
    end
  end

有更好的建议吗？

Answer 1

尝试将整个代码包装到单个数据库事务中。既然你在Heroku，那将是一个Postgres的低端。有了这么多update语句，您可能一次处理这些语句就会大大受益，因此您的代码执行得更快，基本上只留下一个由6500个语句组成的“队列”，以便在Postgres端运行，因为服务器能够对它们进行排队列。根据底层的不同，您可能不得不将事务处理成较小的块--但即使一次处理100次(然后关闭并重新打开事务)也会极大地提高Pg的吞吐量。

http://api.rubyonrails.org/classes/ActiveRecord/Transactions/ClassMethods.html http://www.postgresql.org/docs/9.2/static/sql-set-transaction.html

因此，在第2行之前，需要添加如下内容：

def add_details(shop, shopify_orders)
  Order.transaction do
    shopify_orders.each do |shopify_order|

然后在方法的末尾添加另一个端点：

      if !payment_details.blank?
        PaymentDetail.add_details(order, payment_details)
      end
    end //shopify_orders.each..
  end //Order.transaction..
end //method

Answer 2

您可以像这样对ActiveRecord进行猴子补丁：

class ActiveRecord::Base

  #http://stackoverflow.com/questions/15317837/bulk-insert-records-into-active-record-table?lq=1
  #https://gist.github.com/jackrg/76ade1724bd816292e4e
  #  "UPDATE THIS SET <list_of_column_assignments>  FROM <table_name> THIS  JOIN (VALUES (<csv1>, <csv2>,...) VALS ( <column_names> ) ON <list_of_primary_keys_comparison>"
  def self.bulk_update(record_list)
      pk = self.primary_key
      raise "primary_key not found" unless pk.present?

      raise "record_list not an Array of Hashes" unless record_list.is_a?(Array) && record_list.all? {|rec| rec.is_a? Hash }
      return nil if record_list.empty?

      result = nil

      #test if every hash has primary keys, so we can JOIN
      record_list.each { |r|  raise "Primary Keys '#{self.primary_key.to_s}' not found on record: #{r}" unless hasAllPKs?(r) }


      #list of primary keys comparison
      pk_comparison_array = []
      if (pk).is_a?(Array)
          pk.each {|thiskey| pk_comparison_array << "THIS.#{thiskey} = VALS.#{thiskey}" }
      else
          pk_comparison_array << "THIS.#{pk} = VALS.#{pk}"
      end
      pk_comparison = pk_comparison_array.join(' AND ')

      #SQL
      (1..record_list.count).step(1000).each do |start|
        key_list, value_list = convert_record_list(record_list[start-1..start+999])
        #csv values
        csv_vals = value_list.map {|v| "(#{v.join(", ")})" }.join(", ")
        #column names
        column_names = key_list.join(", ")
        #list of columns assignments
        columns_assign_array = []
        key_list.each {|col|
          unless inPK?(col)
            columns_assign_array << "THIS.#{col} = VALS.#{col}"
          end }
        columns_assign = columns_assign_array.join(', ')

        sql = "UPDATE THIS SET #{columns_assign}  FROM #{self.table_name} THIS  JOIN ( VALUES #{csv_vals} ) VALS ( #{column_names} ) ON ( #{pk_comparison} )"
        result = self.connection.execute(sql)

        return result if result<0
      end

      return result

  end

  def self.inPK?(str)
      pk = self.primary_key

      test = str.to_s
      if pk.is_a?(Array)
            (pk.include?(test))
      else
            (pk==test)
      end
  end

  #test if given hash has primary keys included as hash keys and those keys are not empty
  def self.hasAllPKs?(hash)
      h = hash.stringify_keys
      pk = self.primary_key

      if pk.is_a?(Array)
           (pk.all? {|k| h.key?(k) and h[k].present? })
      else
           h.key?(pk) and h[pk].present?
      end
  end

  def self.convert_record_list(record_list)
    # Build the list of keys
    key_list = record_list.map(&:keys).flatten.map(&:to_s).uniq.sort

    value_list = record_list.map do |rec|
      list = []
      key_list.each {|key| list <<  ActiveRecord::Base.connection.quote(rec[key] || rec[key.to_sym]) }
      list
    end

    # If table has standard timestamps and they're not in the record list then add them to the record list
    time = ActiveRecord::Base.connection.quote(Time.now)
    for field_name in %w(created_at updated_at)
      if self.column_names.include?(field_name) && !(key_list.include?(field_name))
        key_list << field_name
        value_list.each {|rec| rec << time }
      end
    end

    return [key_list, value_list]
  end
end

然后，您可以生成一个包含模型属性(包括它们的主键)的散列数组，并执行如下操作：

ActiveRecord::Base.transaction do
   Model.bulk_update [ {attr1: val1, attr2: val2,...},  {attr1: val1, attr2: val2,...},   ... ]
end

它将是一个没有Rails回调和验证的SQL命令。

Answer 3

对于PostgreSQL，有几个问题上面的方法没有解决：

1. 必须在update目标表中指定实际表，而不仅仅是别名。
2. 不能在FROM短语中重复目标表。由于要将目标表加入值表(因此在FROM短语中只有一个表)，您将无法使用JOIN，所以必须使用"WHERE“。
3. 在一个值表中，您不会得到与在一个简单的"UPDATE“命令中所做的相同的”空闲“强制转换，因此您必须将日期/时间戳值转换为这样的值(#val_cast这样做)。 class ActiveRecord::Base self.update!(record_list)引发ArgumentError "record_list not a Array of Hashes“，除非record_list.is_a？(数组) && record_list.all？{AC.26 rec rec.is_a？？返回record_list )如果record_list.empty？(1..record_list.count).step(1000).each do \start start field_list，value_list = key_field = self.primary_key non_key_fields = field_list - [%Q"#{self.primary_key}"，%Q"created_at"] columns_assign = non_key_fields.map {x{field} "#{field} = #{val_cast(field)}"}.join("，") value_table = value_list.map {{行"(#{row.join("，“}})”}.join("，“)") sql =“更新#{table_name}作为此集合#{columns_assign}从(值#{value_table}) vals (#{field_list.join("，")})开始，其中this.#{key_field} = vals.#{key_field}”self.connection.update_sql(Key_field)结束返回record_list end def self.val_cast(字段)字段=field.gsub(‘’)，'')如果(列=columns.find{x=c\c c.name ==字段}).sql_type =~ /time/“强制转换”(vals.#{ field } as #{column.sql_type})“vals.#{field}”结束vals.# self.convert_record_list(record_list) #构建字段列表field_list = record_list.map(&:keys).flatten.map(&:to )( _s).uniq.sort value_list = record_list.map do \rec\ list = [] field_list.each {\x_~_L_( updated_at)如果self.column_names.include?( field_name ) & !(field_list.include?(field_name)) field_list << field_name value_list.each {AC.26 rec \rec << time } end field_list.map！{AC.26 field %Q"#{field}“}返回field_list，value_list端