Обзор

Мне нужно обновить данные в базе данных, принадлежащей мне, с данными, которые извлекаются с сервера Snowflake. Я извлекаю данные за конкретную дату из Snowflake и нуждаюсь в их вставке в локальную базу данных на эту дату. Эта операция в настоящее время занимает 273 секунды, и я должен оптимизировать ее как можно больше. Все это делается на Python с использованием библиотеки pyodbc для вставки в локальную базу данных и sqlalchemy для извлечения данных из Snowflake.

Подробный процесс

Я использую cProfile в сочетании с Snakeviz для профилирования и визуализации производительности моей операции, которая выполняется для конкретной даты и называется update_snf. Весь процесс состоит из следующих шагов:

1. Создание соединения и выполнение SQL-запроса для извлечения данных с сервера Snowflake. Назовем эту операцию get_data_from_snf. Выполнение этой операции занимает 111 секунд, из которых большинство (108 секунд) занимает вызов функции Pandas.read_sql(query, connection) (определенной в pandas/io/sql.py), которая выполняет запрос и возвращает результат в датафрейме df.
2. Форматирование df, что занимает относительно меньше времени (17.8 секунд)
3. Вставка отформатированного датафрейма df в мою локальную базу данных (назовем эту функцию insert_data_into_db). Эта операция занимает большинство времени (143 секунды, из которых 93.1 секунды занимает метод executemany объекта pyodbc.Cursor и 43.3 секунды методом execute). Назовем таблицу, в которую происходит эта вставка table. table — это довольно большая таблица с примерно 150 миллионами кортежей и 130 столбцами.

Я пытаюсь сократить время выполнения всей этой операции любым возможным способом, оптимизируя любой или даже все шаги. Ожидается, что df будет содержать около 200,000 строк для каждой даты.

Текущая попытка

Используются довольно базовые API Python и в данный момент не используется многопоточность или многопроцессорность. Вставка данных в очень большую table занимала неприемлемое количество времени, поэтому сейчас используется альтернатива, которая реализует insert_data_into_db следующим образом:

1. Создается новая таблица под названием table_temp (эта таблица уже существует в базе данных; она не создается заново каждый раз, когда функция запускается)
2. Для выполнения следующего SQL-запроса используется cursor.execute:

TRUNCATE TABLE table_temp
DELETE FROM table where Date = {date}

Операция удаления необходима для обеспечения того, чтобы только обновленные записи из Snowflake присутствовали на локальном сервере и не было дублирования. Она фактически удаляет устаревшие данные перед тем, как новые данные могут быть вновь вставлены.
3. df разделяется на части по 18000 строк и cursor.fast_executemany устанавливается в True. Теперь мы используем cursor.executemany(insert_query, df_chunk) для вставки каждой из частей (список кортежей) по одной в table_temp.
4. Мы теперь вставляем всю table_temp (которая содержит все записи для конкретной даты из Snowflake) в table используя:

INSERT INTO table
SELECT * FROM table_temp

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

Вопросы в этом текущем подходе:

• Есть ли способ избежать подхода с разбивкой и напрямую вставить в table, который не занимает слишком много времени? Похоже, что время, необходимое для этого прямого вставления, увеличивается по мере того, как table становится больше, и при текущем размере в 150 миллионов строк это занимает слишком много времени.
• Можем ли мы использовать параллелизм в форме многопроцессорности или многопоточности в любом месте этого подхода?
• Можем ли мы использовать стриминг где-то здесь? Если да, то как?
• Можем ли мы ускорить извлечение данных из Snowflake в датафрейм в памяти перед вставкой датафрейма в базу данных? Есть ли способ избежать этого вообще?

Я использую Python 3.11 на сервере Windows 8 с использованием SQL Server. Все оптимизации приветствуются! Большое спасибо!