Мне нужно обновить данные в принадлежащей мне базе данных данными, полученными из базы данных сервера Snowflake. Я получаю данные за определенную дату из Snowflake, и мне нужно вставить их в локальную базу данных на эту дату. В настоящее время эта операция занимает 273 секунды, и мне необходимо максимально ее оптимизировать. Все это делается на Python с использованием библиотеки pyodbc для вставки в локальную базу данных и sqlalchemy для получения данных из Snowflake.
Подробный процесс
Я использую cProfile в сочетании со Snakeviz для профилирования и визуализации производительности моей операции, которая выполняется для определенной даты, называемой update_snf. Весь процесс состоит из следующих шагов:
- Создание соединения и выполнение SQL-запроса для получения данных с сервера Snowflake. Вызовите эту операцию get_data_from_snf. Выполнение этой операции занимает 111 с, причем большую часть (108 с) времени занимает вызов функции Pandas.read_sql(query, Connection). (определено в pandas/io/sql.py), который выполняет запрос и возвращает результат в кадре данных df.
- Форматирование df, что занимает относительно меньше времени (17,8 с)
- Вставка форматированного кадра данных df в мою локальную базу данных (вызовите эту функцию Insert_data_into_db) . Эта операция занимает большую часть времени (143 с, из которых 93,1 с занимает метод выполнения объекта pyodbc.Cursor и 43,3 с методом выполнения). Назовем таблицу, в которую происходит эта вставка, table. table – довольно большая таблица, содержащая примерно 150 миллионов кортежей и 130 столбцов.
Текущая попытка
Довольно простой Python API используются, и многопроцессорность или многопоточность в настоящий момент не используются. Вставка данных в очень большую таблицу занимала неприемлемое количество времени, поэтому в настоящее время используется альтернатива, реализующая вставку_data_into_db следующим образом:
- Создается новая таблица с именем table_temp (эта таблица уже существует в базе данных; она не создается заново при каждом запуске функции)
- используется для выполнения следующего SQL-запроса:
Код: Выделить всё
cursor.execute
Код: Выделить всё
TRUNCATE TABLE table_temp
DELETE FROM table where Date = {date}
3. df разбивается на фрагменты по 18000 строк, а для курсора.fast_executemany установлено значение True. Теперь мы используем курсор.executemany(insert_query, df_chunk), чтобы вставить каждый из фрагментов (список кортежей) один за другим в table_temp.
4. Теперь мы вставляем всю таблицу table_temp (которая содержит все записи для конкретной даты из Snowflake) в таблицу, используя:
Код: Выделить всё
INSERT INTO table
SELECT * FROM table_temp
Вопросы в этом текущем подходе:
- Есть ли какой-либо способ избежать этого подхода к фрагментированию и напрямую вставить в таблицу< /code> что не займет много времени? Похоже, что время, необходимое для этой прямой вставки, увеличивается по мере того, как таблица становится больше, и при текущем размере 150 миллионов строк это занимает слишком много времени.
- Можем ли мы использовать параллелизм в форме многопроцессорности или многопоточности где-нибудь в этом подходе?
- Можем ли мы использовать потоковую передачу где-нибудь здесь? Если да, то как?
- Можем ли мы ускорить извлечение данных из Snowflake в Dataframe в памяти перед вставкой Dataframe в базу данных? Есть ли способ вообще избежать этого?
Подробнее здесь: https://stackoverflow.com/questions/789 ... arge-table
