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

@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 секунды.

Новые блоги и статьи Все статьи Все блоги /
Символьное дифференцирование igorrr37 13.02.2026 / * Программа принимает математическое выражение в виде строки и выдаёт его производную в виде строки и вычисляет значение производной при заданном х Логарифм записывается как: (x-2)log(x^2+2) -. . .	Камера Toupcam IUA500KMA Eddy_Em 12.02.2026 Т. к. у всяких "хикроботов" слишком уж мелкий пиксель, для подсмотра в ESPriF они вообще плохо годятся: уже 14 величину можно рассмотреть еле-еле лишь на экспозициях под 3 секунды (а то и больше),. . .	И ясному Солнцу zbw 12.02.2026 И ясному Солнцу, и светлой Луне. В мире покоя нет и люди не могут жить в тишине. А жить им немного лет.	«Знание-Сила» zbw 12.02.2026 «Знание-Сила» «Время-Деньги» «Деньги -Пуля»
SDL3 для Web (WebAssembly): Подключение Box2D v3, физика и отрисовка коллайдеров 8Observer8 12.02.2026 Содержание блога Box2D - это библиотека для 2D физики для анимаций и игр. С её помощью можно определять были ли коллизии между конкретными объектами и вызывать обработчики событий столкновения. . . .	SDL3 для Web (WebAssembly): Загрузка PNG с прозрачным фоном с помощью SDL_LoadPNG (без SDL3_image) 8Observer8 11.02.2026 Содержание блога Библиотека SDL3 содержит встроенные инструменты для базовой работы с изображениями - без использования библиотеки SDL3_image. Пошагово создадим проект для загрузки изображения. . .	SDL3 для Web (WebAssembly): Загрузка PNG с прозрачным фоном с помощью SDL3_image 8Observer8 11.02.2026 Содержание блога Библиотека SDL3_image содержит инструменты для расширенной работы с изображениями. Пошагово создадим проект для загрузки изображения формата PNG с альфа-каналом (с прозрачным. . .	Установка Qt-версии Lazarus IDE в Debian Trixie Xfce volvo 10.02.2026 В общем, достали меня глюки IDE Лазаруса, собранной с использованием набора виджетов Gtk2 (конкретно: если набирать текст в редакторе и вызвать подсказку через Ctrl+Space, то после закрытия окошка. . .

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

	Обработка больших данных и сохранение их в базу данных 21.08.2020, 11:10. Показов 4099. Ответов 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

Рыжий Лис Просто Лис 5973 / 3735 / 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