MySQL批量插入多个表

Posted 2023-05-09

【中文标题】MySQL批量插入多个表

【英文标题】：MySQL bulk insert on multiple tables

【发布时间】：2020-02-04 20:12:08

【问题描述】：

我有一个 mysql 数据库，其中包含 2 个表 products 和 product_variants。一个产品有许多产品变体。这是一个示例：

products
+----+------+
| id | name |
+----+------+
|  1 | Foo  |
|  2 | Bar  |
+----+------+

product_variants
+----+-------------+--------+
| id | product_id  | value  |
+----+-------------+--------+
| 10 |           1 | red    |
| 11 |           1 | green  |
| 12 |           1 | blue   |
| 13 |           2 | red    |
| 14 |           2 | yellow |
+----+-------------+--------+

现在我需要以最有效和最快的方式批量插入许多产品及其变体。我有一个包含许多产品（100k+）的 JSON，如下所示：

[
  
    "name": "Foo",
    "variants": [ "value": "red" ,  "value": "green" ,  "value": "blue" ]
  ,
  
    "name": "Bar",
    "variants": [ "value": "red" ,  "value": "yellow" ]
  ,
  ...
]

我应该从中生成查询以插入产品。

我的想法是像这样使用insert 查询：

INSERT INTO `products` (name) VALUES ("foo"), ("bar"), ...;

但是我不知道在product_variants的插入查询中使用什么product_id（外键）：

INSERT INTO `product_variants` (product_id,value) VALUES (?,"red"), (?,"green"), ...;

（事务中的这些查询）

我曾考虑手动指定产品 ID，以从最后一个 ID 递增的方式，但当并发连接同时插入产品或同时运行 2 个或多个批量插入进程时，我会收到错误。

我可以使用什么策略来实现我的目标？有标准的方法吗？

ps：如果可能的话，我不想改变这两个表的结构。

【问题讨论】：

如果您坚持使用这样的插入，则需要在为“Foo”构建 product_variants 插入之前执行 SELECT id FROM products WHERE name = "Foo"，然后根据需要为每个附加产品执行另一个插入。也就是说，在许多环境中，我发现使用更简单的预准备语句进行插入具有与诸如此类的“批量”插入相当的性能。

【参考方案1】：

您可以使用last_insert_id() 从最后一条语句中获取最后生成的 ID。但是，如上所述，由于这仅获取语句的最后一个 ID，因此需要您单独处理每个产品。您可以批量插入变体。但是从给定 JSON 的结构来看，我认为这使得遍历该 JSON 变得更加容易。每个产品及其变体都应插入事务中，这样如果产品表中的INSERT 由于某种原因失败，产品的变体就不会添加到先前的产品中。

START TRANSACTION;
INSERT INTO products
            (name)
            VALUES ('Foo');
INSERT INTO product_variants
            (product_id,
             value)
            VALUES (last_insert_id(),
                    'red'),
                   (last_insert_id(),
                    'green'),
                   (last_insert_id(),
                    'blue');
COMMIT;

START TRANSACTION;
INSERT INTO products
            (name)
            VALUES ('Bar');
INSERT INTO product_variants
            (product_id,
             value)
            VALUES (last_insert_id(),
                    'red'),
                   (last_insert_id(),
                    'yellow');
COMMIT;

db<>fiddle

【讨论】：

应该将查询放入事务BEGIN; INSERT ...; INSERT ...; COMMIT;

【参考方案2】：

如果您已经将 JSON 保存在一个表中，那么可以通过两条语句（非常有效地）完成：

INSERT INTO Products (name)
    SELECT name
        FROM origial_table;  -- to get the product names

INSERT INTO Variants (product_id, `value`)
    SELECT  ( SELECT id FROM Products WHERE name = ot.name ),
            `value`
        FROM origial_table AS ot;

实际上，name 和 value 需要是合适的 JSON 表达式才能提取值。

如果您担心第一张表中有大量重复的“产品”，请务必使用UNIQUE(name)。您可以通过此处描述的两步过程避免“燃烧”ID：mysql.rjweb.org/doc.php/staging_table#normalization

【参考方案3】：

最后，我使用了一种使用 MySQL 函数 LAST_INSERT_ID() 的策略，例如 @sticky-bit sad，但使用更快的批量插入（许多产品为 1 次插入）。

我附上了一个简单的 Ruby 脚本来执行批量插入。似乎一切都适用于并发插入。

我已经运行带有标志innodb_autoinc_lock_mode = 2 的脚本，一切似乎都很好，但我不知道是否有必要将标志设置为 1：

require 'active_record'
require 'benchmark'
require 'mysql2'
require 'securerandom'

ActiveRecord::Base.establish_connection(
  adapter:  'mysql2',
  host:     'localhost',
  username: 'root',
  database: 'test',
  pool:     200
)

class ApplicationRecord < ActiveRecord::Base
  self.abstract_class = true
end

class Product < ApplicationRecord
  has_many :product_variants
end

class ProductVariant < ApplicationRecord
  belongs_to :product
  COLORS = %w[red blue green yellow pink orange].freeze
end

def migrate
  ActiveRecord::Schema.define do
    create_table(:products) do |t|
      t.string :name
    end

    create_table(:product_variants) do |t|
      t.references :product, null: false, foreign_key: true
      t.string :color
    end
  end
end

def generate_data
  d = []
  100_000.times do
    d << 
      name: SecureRandom.alphanumeric(8),
      product_variants: Array.new(rand(1..3)).map do
         color: ProductVariant::COLORS.sample 
      end
    
  end
  d
end

DATA = generate_data.freeze

def bulk_insert
  # All inside a transaction
  ActiveRecord::Base.transaction do
    # Insert products
    values = DATA.map  |row| "('#row[:name]')" .join(',')
    q = "INSERT INTO products (name) VALUES #values"
    ActiveRecord::Base.connection.execute(q)

    # Get last insert id
    q = 'SELECT LAST_INSERT_ID()'
    last_id, = ActiveRecord::Base.connection.execute(q).first

    # Insert product variants
    i = -1
    values = DATA.map do |row|
      i += 1
      row[:product_variants].map  |subrow| "(#last_id + i,'#subrow[:color]')" 
    end.flatten.join(',')
    q = "INSERT INTO product_variants (product_id,color) VALUES #values"
    ActiveRecord::Base.connection.execute(q)
  end
end

migrate

threads = []

# Spawn 100 threads that perform 200 single inserts each
100.times do
  threads << Thread.new do
    200.times do
      Product.create(name: 'CONCURRENCY NOISE')
    end
  end
end

threads << Thread.new do
  Benchmark.bm do |benchmark|
    benchmark.report('Bulk') do
      bulk_insert
    end
  end
end

threads.map(&:join)

运行脚本后，我检查了所有产品是否都有与查询相关联的变体

SELECT * 
FROM products
 LEFT OUTER JOIN product_variants
 ON (products.id = product_variants.product_id)
WHERE product_variants.product_id IS NULL
AND name != "CONCURRENCY NOISE";

而且我没有得到任何行。