Интересное что-то

Forwarded from Инженерообязанный🫡 | Блог Дата Инженера

#кейсы

🚨Ребят, рассказываю про кейс, с которым столкнулся🚨.

Вероятнее всего будет интересен Аналитикам из-за работы с Pandas, но Инженерам данных тоже к прочтению.🧑‍🎓

Вкратце, есть Дата Фрейм, и в нём 12к строк, и 22 колонки, и джоба работает абсолютно нормально, все преобразования работают за секунду, а заливка данных в GreenPlum 13 минут работает. Сами понимаете,12к строк всего, а если там будет 100к строк? Что тогда🤔? (Считать не нужно, вопрос риторический)

Пример кода загрузки в GreenPlum:

def df_to_db_old(df, schema, table, force=False):
    from sqlalchemy import text
    engine = get_gp_engine() # get_gp_engine функция возвращающая движок от psycopg3
    df.to_sql(con=engine,
              index=False,
              schema=schema,
              name=table,
              if_exists='append',
              chunksize=16000)
# Создание DataFrame
df_5000_rows = pd.DataFrame(data)

df_to_db(df=df_5000_rows, schema='team_test', table='shust_test')

Оказывается, что при использовании стандартной функции to_sql, данные записываются в базу данных построчно, грубо говоря командой INSERT, а параметр chunksize задаёт через сколько строк сделать коммит в БД. При таком использовании время вставки 5к строк составляет:

Время вставки (1): 190.558109998703 секунд

Нам конечно же такая тема не понравилась, так как очень долго. Начал думать🤔, что и как сделать так, чтобы ускорить данный процесс. В голову приходила только идея💡 сохранять данные в файл и использовать команду COPY. И в итоге, нарвался на статью от SQL-EX(ссылка). В общем и целом, там рассматривается 5 способов загрузки данных с помощью библиотеки Pandas в Postgres(если кто не знал 😳 для GreenPlum используется движок Postgresql, поэтому данные варианты подходили и нам). Выбрав самый быстрый способ🐆, оказалось, что в параметр method можно передать свою функцию, где мы будет передавать логику сохранения файла(в нашем случае все сохраняется в памяти) и выполнения команды COPY, и что не мало важно, код был понятным для аналитиков. Код получился следующим:

def df_to_db(df, schema, table, force=False):

    def psql_insert_copy(table, conn, keys, data_iter): #функция передаваемая в мотод
        from io import StringIO
        import csv
       # получаем подключение DBAPI, которое может обеспечить курсор
        dbapi_conn = conn.connection
        with dbapi_conn.cursor() as cur:
            s_buf = StringIO()
            writer = csv.writer(s_buf)
            writer.writerows(data_iter)
            s_buf.seek(0)

            columns = ', '.join('"{}"'.format(k) for k in keys)
            if table.schema:
                table_name = '{}.{}'.format(table.schema, table.name)
            else:
                table_name = table.name

            sql = 'COPY {} ({}) FROM STDIN WITH CSV'.format(
                table_name, columns)
            cur.copy_expert(sql=sql, file=s_buf)

    engine = gp_engine() # gp_engine функция возвращающая движок от psycopg2
    df.to_sql(con=engine,
              index=False,
              schema=schema,
              name=table,
              if_exists='append',
              method=psql_insert_copy)

df_to_db(df=df_5000_rows, schema='team_test', table='shust_test')

В данном случае - время резко сократилось в +100500 раз, и чем больше данных, тем данный метод будет быстрее.

Время вставки (2): 0.3825254440307617 секунд

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

‼️Могу посоветовать сохранить к себе данную статью, чтобы при столкновении с подобной задачей, долго не искать как и что.(Для этого я его тут и сохранил😂 ).

Интересно, а GPT напишет такой код? Хмммм....

Please open Telegram to view this post

VIEW IN TELEGRAM

49 views10:24