Как ускорить импорт из csv в бд?

Question

[vikholodov]

Имеем django, postgresql и 15 миллионов товаров в файле csv, все это импортируется в 1 поток.
Примерно так это делаю:

with open('new_products.csv', 'r', encoding='utf-8') as csvfile:
...
for row in rows:
    product = Product()
    product.price = row[0]
    product.shipping = row[1]
    product.save()

и так далее. Возможно есть какой-то способ ускорить этот процесс в несколько раз?

Answer 1

[neatsoft]

1. Итерировать строки по несколько сотен штук, и создавать соответствующие им объекты единым вызовом к bulk_create:

import itertools

def grouper(iterable, n, fillvalue=None):
    args = [iter(iterable)] * n
    return itertools.zip_longest(*args, fillvalue=fillvalue)

...

limit = 250

for index, items in enumerate(grouper(rows, limit)):
    items = (i for i in items if i) # remove empty rows added by grouper

    products = []

    for item in items:
        product = Product(
            price=item[0],
            shipping=item[1],
        )
        products.append(product)

    Product.objects.bulk_create(products)

Все 15 миллионов объектов разом создавать не стоит - ничего хорошего из этого не выйдет.

2. Если необходимо создавать зависимые объекты, использовать транзакции:

from django.db import transaction

...

limit = 250

for index, items in enumerate(grouper(rows, limit)):
    with transaction.atomic():
        for item in (i for i in items if i):
            product = Product.objects.create(
                price=item[0],
                shipping=item[1],
            )
            # product can be used there to create another objects, e.g.:
            for color in item[2].split(','):
                ProductColor.objects.create(
                    product=product,
                    color=color,
                )

Транзакции, в данном случае, обеспечивают существенный прирост производительности.

3. Запустить несколько процессов, каждому из которых передать в параметрах с какой позиции он должен начать, и сколько строк обработать. Чтобы дочерним процессам не пришлось читать весь файл с самого начала, позицию лучше передавать не номером строки, а смещением.

4. Скомбинировать 3 и 1 или 3 и 2.

5. Модели в джанго - это просто таблицы в базе данных. Можно добавлять строки с помощью SQL запросов из отдельного скрипта на Python или Go.

Comments

[neatsoft]

items = (i for i in items if i) можно поместить в сам grouper:

def grouper(iterable, n):
    args = [iter(iterable)] * n
    groups = itertools.zip_longest(*args, fillvalue=None)
    for group in groups:
        yield (item for item in group if item)

[vikholodov]

Спасибо за очень подробный ответ, сейчас пытаюсь реализовать, но не могу понять как добавить m2m поле. Раньше я сохранял объект и добавлял ему

product.category_id.add(Category.objects.get(catId=str(category_id)))

[neatsoft]

vikholodov,
Метод add добавляет строку в таблицу, хранящую m2m связи. К этой таблице можно обращаться с помощью through. Например, если есть модель Category и модель Product с m2m полем categories, то к эта таблица будет доступна через Product.categories.through. Но нужно учитывать, что на момент добавления строк в m2m таблицу должны быть известны и category_id, и product_id.

При использовании второго варианта ускорения (оборачивание в транзакцию) ничего дополнительно предпринимать не нужно, можно применять и add, и пример из моего ответа.

Если используется Django >1.10 и PostgreSQL, можно задействовать и первое решение, т.к. bulk_update в такой конфигурации возвращает список айдишников созданных объектов. Нужно просто сохранять в отдельном списке категории, в которые должен оказаться включен каждый из продуктов, и ещё одним вызовом bulk_create создавать строки в таблице с m2m связями:

product_category_ids = []

# create products
products = []

for item in items:
    product = Product(
        price=item[0],
        shipping=item[1],
    )
    products.append(product)
    category_ids = ... # get category ids from the input data
    product_category_ids.append(category_ids)

product_ids = Product.objects.bulk_create(products)

# create m2m relations between products and categories
product_categories = []

for product_id, category_ids in zip(product_ids, product_category_ids):
    for category_id in category_ids:
        product_category = Product.categories.through(
            product_id=product_id,
            category_id=category_id,
        )
        product_categories.append(product_category)

Product.categories.through.objects.bulk_create(product_categories)

[vikholodov]

Столкнулся с ошибкой:

django.db.utils.IntegrityError: duplicate key value violates unique constraint "core_product_sku_key"
DETAIL:  Key (sku)=(6701154907) already exists.

Пробовал уже тысячу проверок различных делать (наличие в базе, наличие в списке products), не помогает. + фактически этого самого ключа нет в базе! 2.5 млн товаров добавил и споткнулся вот на этой позиции

[neatsoft]

vikholodov, Несколько продуктов с одинаковым ключом в исходном списке?
Либо фильтровать (используя в качестве фильтра не только строки из базы, но и из списка - если есть несколько продуктов с одинаковым айдишником, добавляться должен только один из них), либо заменить create на update_or_create / get_or_create.