Распараллеливание запросов в PostgreSQL с помощью Python

В Geoblink мы выполняем более 20000 запросов для создания одной из наших нескольких баз данных PostgreSQL размером ~ 100 ГБ с нуля из файлов необработанных данных. Если мы попытаемся запустить их в последовательном порядке, создание базы данных займет слишком много времени. Чтобы сократить время генерации, мы распараллеливаем несколько запросов. Сделать это вручную было бы невозможно, поэтому мы используем хороший скрипт, написанный на Python, для генерации и выполнения запросов.

В этом посте я покажу пример того, как это сделать на Python.

В качестве примера мы создадим таблицу и заполним ее 20 миллионами случайных чисел.

Сначала создаем таблицу:

DROP TABLE IF EXISTS public.test; 
CREATE TABLE public.test( value numeric );

А потом заливаем:

INSERT INTO test 
SELECT random() 
FROM generate_series(1,20000000);

На моей машине (процессор Intel (R) Core (TM) i7–4790 @ 3,60 ГГц) это занимает около 64 секунд.

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

Сначала мы определяем функцию для запуска запроса:

import psycopg2 
def runQuery(query): 
  connect_text = "dbname='%s' user='%s' host=%s port=%s password='%s'" % (dbname, user, host, port, password) 
  con = psycopg2.connect(connect_text) 
  cur = con.cursor() 
  cur.execute(query) 
  con.commit() 
  con.close()

С помощью этой функции мы можем выполнять любой запрос, но без вывода, например создание таблиц, их обновление и т. Д.
Следующим шагом является создание массива запросов.

queries = [] 
for i in range(N_job): 
  query = """ 
    INSERT INTO test (value) 
    SELECT random() 
    FROM generate_series(1, %d); 
  """ % (N_random / N_job) 
query.append(queries)

Где N_jobs - количество запросов, которые мы хотим выполнить. Таким образом, мы создали массив независимых запросов, которые можем выполнять параллельно.
Теперь нам нужно создать пул с количеством процессоров, которые мы хотим использовать, в нашем случае 4.

import multiprocessing 
pool = multiprocessing.Pool(CPUS)

Наконец, мы выполняем запросы, используя метод imap_unordered. Он принимает функцию и массив и применяет функцию к каждому элементу массива в случайном порядке.

for i in pool.imap_unordered(runQuery, queries) 
  continue

С помощью этого метода расчет занимал всего 30 секунд с использованием 4 ядер, что составляет более 50% улучшения по сравнению с использованием только одного запроса. Вероятно, мы не получили большого улучшения из-за скорости чтения / записи диска.

Чем больше ядер доступно, тем больше преимуществ мы можем извлечь из распараллеливания запросов. Хорошей идеей является распараллеливание с использованием существующих индексированных столбцов в качестве провинций, муниципалитетов или расстояний, как мы это делаем при создании наших баз данных в Geoblink.

Но есть момент, когда мы должны играть с соотношением строк / задание, поскольку мы не хотим иметь миллионы строк на запросы и тысячи запросов.

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

By Vicente "Tito" Lacuesta