Обработка больших данных и сохранение их в базу данных

@m0nte-cr1st0 · Регистрация: 15.01.2019

Студворк — интернет-сервис помощи студентам

Есть веб-сервис на Python 3.7 и Flask 1.1
Флоу следующее:
1. Пользователь может загружать файлы (xls/csv) либо давать урл на гугл таблицу.
2. В таблице может быть неограниченное число строк - до 80к, например.
3. В каждой строке хранятся данные. Из каждой строки создаётся новая запись в базе данных (postgresql)
4. Каждая запись при сохранении индексируется в ElasticSearch.
5. Перед сохранением записи в бд нужно циклом пройти по каждой строке, чтобы привести данные в нужный формат.

Всё это дело я вынес в celery. Но очень уж долго происходит вся эта обработка и сохранение данных.
Как можно оптимизировать? Потоки? Мультипроцессы? Селери чанки?

P.S. Вопрос на SO - https://stackoverflow.com/ques... e-database

@Garry Galler · 21.08.2020, 18:52

Сообщение от m0nte-cr1st0

Потоки? Мультипроцессы? Селери чанки?

На а bottleneck в чем? Нужно же сначала определить, что будет оптимизироваться, а потом уже подбирать механизм.
То есть найти самую медленную операцию: разбор xls или запись в БД или еще что.
Потоки тут только ожидание IO могут скрасить. Поэому или async (дешевле потоков) или процессы, если можно четко разделить данные по таскам и они будут независимы друг от друга.

Добавлено через 5 минут
P.S. Хотя я бы все на Go переписал... Там вообще все летает :-) Потоков нет. Мультипроцессов нет.
Есть только горутины... и их можно хоть миллион создавать... распараллеливание на ядрах прямо из коробки.

@m0nte-cr1st0 · 22.08.2020, 09:34 **[ТС]**

Garry Galler, ну, ну разбираться с go ради одной таски - ну не очень хочется.
Запись в бд происходит быстро. Вся суть в том, что таких записей очень много. И нужно по каждой из них пройти циклом для форматирования входных данных.
Я думал, как-то разбить на чанки эти данные (по строкам) и дальше асинхронно, параллельноих обрабатывать и ложить в бд. Но пока не пойму, как это сделать и какой инструмент взять.

@Garry Galler · 22.08.2020, 13:27

Сообщение от m0nte-cr1st0

И нужно по каждой из них пройти циклом для форматирования входных данных.

Значит нужно разделить процессы обработки xls\csv и запись в БД.
На примере модели которую я использую в Go:
Есть канал (в терминах Python - очередь заданий), который принимает готовые для записи в БД таски.
Есть горутина (поток\ или асинхронная нить в терминах Python), которая эти таски читает и просто пишет в БД (горутина одна, потому что как показала практика БД тип MySQL очень не любят множественные запросы на запись из разных потоков, создается очередь и потом часть запросов начинает отвергаться).
Есть множество горутин, которые параллельно обрабатывают данные от каждого пользователя (файлы xls\csv - причем обработку каждого файла можно дополнительно делить на чанки в субгорутинах) и кидают их в канал (очередь) в виде какой-то готовой для записи структурки.

Главный вопрос: как разделить на чанки обработку отдельного файла, поскольку они большие, то одной горутины (нити) будет маловато. В Go я мог бы сделать отдельную горутину на каждую строчку файла (80 тыс горутин на файл).
Но для Python это, возможно, будет многовато. НО почему бы не попробовать? Разумеется нужно использовать не потоки, а асинхронные нити. А вместо стандартного цикла обработки событий в asyncio поставить более производительный uvloop (event loop).

Рыжий Лис · 22.08.2020, 14:03

sqlalchemy асинхронная запись в базу

@Garry Galler · 22.08.2020, 15:05

Сообщение от Рыжий Лис

sqlalchemy асинхронная запись в базу

А где там именно асинхронная запись?

Что касается бенчмарков - они все идут лесом, если ты вставляешь в БД не строчки на пару десятков байт, а mb-ные данные.

Рыжий Лис · 22.08.2020, 15:16

Это просто название темы

Сообщение от Garry Galler

mb-ные данные.

Что это?

Добавлено через 44 секунды
А, мегабайты... почитай-таки пост по ссылке

@Garry Galler · 22.08.2020, 17:31

Сообщение от Рыжий Лис

почитай-таки пост по ссылке

Там бечмаркаются hello world'ы.
Попробуй сделать бенч на реальных данных.

Добавлено через 6 минут
А executemany можно делать и без sqlalchemy.
Так что смысла к отсылке в эту тему не вижу абсолютно никакой.
Вопрос ТС не в этом. Он и сам знает как bulk-ать\батчить - с ORM или без, благо не новичок.

Добавлено через 7 минут
P.S. А вот попробовать asyncpg vs aiopg ТС'у стоит.

Рыжий Лис · 22.08.2020, 17:57

Не знаю причём здесь асинхронщина, вот на SO тоже предлагают импортировать данные через COPY.

SQL
1
COPY my_table (col1, col2, etc) FROM STDIN WITH CSV

Python
1
2
3
4
5
6
7
8
conn = psycopg2.connect(DSN)
curs = conn.cursor()
 
data = StringIO.StringIO()
data.write('\n'.join(['1', '2','3']))
data.seek(0)
 
curs.copy_from(data, 'my_table')

Люди намерили, что миллион записей вставилось за 82 секунд.

Добавлено через 1 минуту
https://www.trineo.com/blog/20... large-csvs

@m0nte-cr1st0 · 23.08.2020, 10:01 **[ТС]**

Garry Galler,

[Значит нужно разделить процессы обработки xls\csv и запись в БД.

У меня логика обработки файлов и записи в бд разделена на отдельные классы / методы.

Провел замеры на файле в 4.5к строк.
Обработка его, приведение к нужному формату занимает около секунды. Понятно, что на файле в 80к строк это будет занимать больше времени, но тут сложно будет распараллелить, ибо есть связанные строки (например, есть 200 строк с одинаковым городом, город указывается только в одной ячейке, на самой первой строке, остальные 199 - пустые). Поэтому тут либо велосипед какой-то придумывать (в итоге, должно стать быстрее, если все же получится), либо обрабатывать последовательно в 80к итераций...
Делаю запись в бд через add_all (алхимия) - 40-45 секунд уходит. Булк инсерт не подходит, так как есть связи mtm.
Пойду почитаю об асинхронной записи в бд ещё.

Рыжий Лис,

Люди намерили, что миллион записей вставилось за 82 секунд.

Очень интересно. Можно попробовать сохранить файл (в памяти) с уже отформатированными мне данными и вставить его в бд. Спасибо.

Добавлено через 4 минуты
Рыжий Лис, все же, если там под капотом лежит булк инсерт, а видимо, так и есть, то не прокатит, ибо он игнорит mtm

Добавлено через 13 минут
То есть. Есть таблица Article, есть Tag. Между ними - mtm.
Из строк файла создаю записи Article.
В строках есть колонка Tags, где указывается title тэга.
В таблице Tag есть также колонка name, которая генерится на основе title.
Нужно, чтобы при создании Article из файла, создавались также записи в mtm таблице. Если такого тэга нет - создавать его, при этом генерить для него name по определённой логике, в зависимости от title, который указан в файле.

Реальный кейс вообще?

Рыжий Лис · 23.08.2020, 10:31

Сообщение от m0nte-cr1st0

все же, если там под капотом лежит булк инсерт,

Нет, под капотом лежит COPY.

@m0nte-cr1st0 · 23.08.2020, 10:34 **[ТС]**

Нет, под капотом лежит COPY.

Все же вряд ли он создаёт mtm

Рыжий Лис · 23.08.2020, 10:36

Сообщение от m0nte-cr1st0

Реальный кейс вообще?

Я-то думал у тебя там куча связей и хотел предложить сначала заинсёртить справочники - потом следующий "уровень" таблиц.

Нет ничего невозможного. Предлагаю так: любым способом импортируешь таблицу Article. А потом в цикле пробегаешь по всем строкам, у которых нет тегов (читай - нет id в промежуточной таблице) и расставляешь теги/создаёшь новые.

@m0nte-cr1st0 · 23.08.2020, 10:43 **[ТС]**

Рыжий Лис, тоже к этому склоняюсь уже. Но как-то итерировать по 80к строкам - не очень хорошо. Хотя тут уже можно и асинхронщину или процессы порубить ведь?

@m0nte-cr1st0 · 23.08.2020, 20:09 **[ТС]**

Garry Galler, Рыжий Лис, если интересно, то вот, как получается.

Python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
def forming_data_for_db_insert(rows_list: list) -> List[dict]:
    """
    Читает файл построчно, приводит данные к нужному формату.
    Возвращает список словарей, готовый для вставки в бд.
    """
    dict_list = []
    for ind, row in enumerate(rows_list):
        data = self._read(row, ind)
        dict_list.append(data)
    return dict_list
 
 
def articles_create_from_dictionaries(data: list) -> None:
    """Создаёт записи в бд. Преобразовывает связанные объекты в массив инстансов"""
    articles = []
    for article_data in data:
        article_data["categories"] = forming_relation_list_for_article(
            article_data, "categories", CategoryModel
        )
        article_data["tags"] = forming_relation_list_for_article(
            article_data, "tags", TagModel
        )
        articles.append(ArticleModel(**article_data))
    db.session.add_all(articles)
    commit_session()

Потестил на csv размером в 4,5к строк.
Обработка файла, приведение к нужному формату - занимает около секунды. В принципе, вполне прилично.
Запись в бд 4,5к строк, + мтм таблицы - 68 секунд.

Если как-то оптимизировать дальше - нужно работать над articles_create_from_dictionaries, очевидно. Но как? Ведь это запись в бд. Можно ли писать в бд параллельно и есть ли в этом какие риски (транзакционность, мб?)? Не знаю...

Добавлено через 4 минуты
Если бы не было mtm, просто вставить файл в бд, то 4,5к записей вставить можно за 3,5 секунды.

Новые блоги и статьи Все статьи Все блоги /
Советы по крайней бережливости. Внимание, это ОЧЕНЬ длинный пост. Programma_Boinc 28.12.2025 Советы по крайней бережливости. Внимание, это ОЧЕНЬ длинный пост. Налог на собак: https:/ / ********/ gallery/ V06K53e Финансовый отчет в Excel: https:/ / ********/ gallery/ bKBkQFf Пост отсюда. . .	Кто-нибудь знает, где можно бесплатно получить настольный компьютер или ноутбук? США. Programma_Boinc 26.12.2025 Нашел на реддите интересную статью под названием Anyone know where to get a free Desktop or Laptop? Ниже её машинный перевод. После долгих разбирательств я наконец-то вернула себе. . .	Thinkpad X220 Tablet — это лучший бюджетный ноутбук для учёбы, точка. Programma_Boinc 23.12.2025 Рецензия / Мнение/ Перевод Нашел на реддите интересную статью под названием The Thinkpad X220 Tablet is the best budget school laptop period . Ниже её машинный перевод. Thinkpad X220 Tablet —. . .	PhpStorm 2025.3: WSL Terminal всегда стартует в ~ and_y87 14.12.2025 PhpStorm 2025. 3: WSL Terminal всегда стартует в ~ (home), игнорируя директорию проекта Симптом: После обновления до PhpStorm 2025. 3 встроенный терминал WSL открывается в домашней директории. . .	Как объединить две одинаковые БД Access с разными данными VikBal 11.12.2025 Помогите пожалуйста !! Как объединить 2 одинаковые БД Access с разными данными.
Новый ноутбук volvo 07.12.2025 Всем привет. По скидке в "черную пятницу" взял себе новый ноутбук Lenovo ThinkBook 16 G7 на Амазоне: Ryzen 5 7533HS 64 Gb DDR5 1Tb NVMe 16" Full HD Display Win11 Pro	Музыка, написанная Искусственным Интеллектом volvo 04.12.2025 Всем привет. Некоторое время назад меня заинтересовало, что уже умеет ИИ в плане написания музыки для песен, и, собственно, исполнения этих самых песен. Стихов у нас много, уже вышли 4 книги, еще 3. . .	От async/await к виртуальным потокам в Python IndentationError 23.11.2025 Армин Ронахер поставил под сомнение async/ await. Создатель Flask заявляет: цветные функции - провал, виртуальные потоки - решение. Не threading-динозавры, а новое поколение лёгких потоков. Откат?. . .	Поиск "дружественных имён" СОМ портов Argus19 22.11.2025 Поиск "дружественных имён" СОМ портов На странице: https:/ / norseev. ru/ 2018/ 01/ 04/ comportlist_windows/ нашёл схожую тему. Там приведён код на С++, который показывает только имена СОМ портов, типа,. . .	Сколько Государство потратило денег на меня, обеспечивая инсулином. Programma_Boinc 20.11.2025 Сколько Государство потратило денег на меня, обеспечивая инсулином. Вот решила сделать интересный приблизительный подсчет, сколько государство потратило на меня денег на покупку инсулинов. . . .

@m0nte-cr1st0 1043 / 578 / 242 Регистрация: 15.01.2019 Сообщений: 2,178 Записей в блоге: 1

	Обработка больших данных и сохранение их в базу данных 21.08.2020, 11:10. Показов 4062. Ответов 14 Метки celery, flask, postgresql, python (Все метки) Есть веб-сервис на `Python 3.7` и `Flask 1.1` Флоу следующее: 1. Пользователь может загружать файлы (`xls/csv`) либо давать урл на `гугл таблицу`. 2. В таблице может быть неограниченное число строк - до 80к, например. 3. В каждой строке хранятся данные. Из каждой строки создаётся новая запись в базе данных (`postgresql`) 4. Каждая запись при сохранении индексируется в `ElasticSearch`. 5. Перед сохранением записи в бд нужно циклом пройти по каждой строке, чтобы привести данные в нужный формат. Всё это дело я вынес в celery. Но очень уж долго происходит вся эта обработка и сохранение данных. Как можно оптимизировать? Потоки? Мультипроцессы? Селери чанки? P.S. Вопрос на SO - https://stackoverflow.com/ques... e-database 0

@m0nte-cr1st0 1043 / 578 / 242 Регистрация: 15.01.2019 Сообщений: 2,178 Записей в блоге: 1
	22.08.2020, 09:34 [ТС]
	Garry Galler, ну, ну разбираться с go ради одной таски - ну не очень хочется. Запись в бд происходит быстро. Вся суть в том, что таких записей очень много. И нужно по каждой из них пройти циклом для форматирования входных данных. Я думал, как-то разбить на чанки эти данные (по строкам) и дальше асинхронно, параллельноих обрабатывать и ложить в бд. Но пока не пойму, как это сделать и какой инструмент взять. 0

Рыжий Лис Просто Лис 5972 / 3734 / 1099 Регистрация: 17.05.2012 Сообщений: 10,791 Записей в блоге: 9
	22.08.2020, 14:03
	sqlalchemy асинхронная запись в базу 0

@m0nte-cr1st0 1043 / 578 / 242 Регистрация: 15.01.2019 Сообщений: 2,178 Записей в блоге: 1
	23.08.2020, 10:43 [ТС]
	Рыжий Лис, тоже к этому склоняюсь уже. Но как-то итерировать по 80к строкам - не очень хорошо. Хотя тут уже можно и асинхронщину или процессы порубить ведь? 0