Как записать данные в Redshift, которые являются результатом фрейма данных, созданного в Python?

Вы можете использовать to_sql для отправки данных в базу данных Redshift. Я смог сделать это, используя соединение с моей базой данных через механизм SQLAlchemy. Просто не забудьте установить index = False в вызове to_sql. Таблица будет создана, если она не существует, и вы можете указать, хотите ли вы, чтобы вызов заменял таблицу, добавлял к таблице или терпел неудачу, если таблица уже существует.

from sqlalchemy import create_engine
import pandas as pd

conn = create_engine('postgresql://username:[email protected]:5439/yourdatabase')

df = pd.DataFrame([{'A': 'foo', 'B': 'green', 'C': 11},{'A':'bar', 'B':'blue', 'C': 20}])

df.to_sql('your_table', conn, index=False, if_exists='replace')

Обратите внимание, что вам может потребоваться pip install psycopg2, чтобы подключиться к Redshift через SQLAlchemy.

документация по_sql

import pandas_redshift as pr

pr.connect_to_redshift(dbname = <dbname>,
                        host = <host>,
                        port = <port>,
                        user = <user>,
                        password = <password>)

pr.connect_to_s3(aws_access_key_id = <aws_access_key_id>,
                aws_secret_access_key = <aws_secret_access_key>,
                bucket = <bucket>,
                subdirectory = <subdirectory>)

# Write the DataFrame to S3 and then to redshift
pr.pandas_to_redshift(data_frame = data_frame,
                        redshift_table_name = 'gawronski.nba_shots_log')

Подробности: https://github.com/agawronski/pandas_redshift

Я пытался использовать pandas df.to_sql(), но это было очень медленно. Мне потребовалось более 10 минут, чтобы вставить 50 строк. См. эту открытую проблему (на момент написания)

Я пытался использовать odo из экосистемы blaze (согласно рекомендациям в обсуждении проблемы), но столкнулся с ProgrammingError который я не удосужился исследовать.

Наконец, что сработало:

import psycopg2

# Fill in the blanks for the conn object
conn = psycopg2.connect(user = 'user',
                              password = 'password',
                              host = 'host',
                              dbname = 'db',
                              port = 666)
cursor = conn.cursor()

# Adjust ... according to number of columns
args_str = b','.join(cursor.mogrify("(%s,%s,...)", x) for x in tuple(map(tuple,np_data)))
cursor.execute("insert into table (a,b,...) VALUES "+args_str.decode("utf-8"))

cursor.close()
conn.commit()
conn.close()

Да, старый добрый psycopg2. Это для массива numpy, но преобразование из df в ndarray не должно быть слишком сложным. Это дало мне около 3 тыс. строк в минуту.

Тем не менее, самое быстрое решение в соответствии с рекомендациями других товарищей по команде — использовать команду COPY после сброса фрейма данных в виде TSV/CSV в кластер S3 и последующего копирования. Вам следует изучить это, если вы копируете действительно огромные наборы данных. (Я обновлю здесь, если и когда попробую)

Раньше я полагался на функцию pandas to_sql(), но она слишком медленная. Недавно я переключился на следующие действия:

import pandas as pd
import s3fs # great module which allows you to read/write to s3 easily
import sqlalchemy

df = pd.DataFrame([{'A': 'foo', 'B': 'green', 'C': 11},{'A':'bar', 'B':'blue', 'C': 20}])

s3 = s3fs.S3FileSystem(anon=False)
filename = 'my_s3_bucket_name/file.csv'
with s3.open(filename, 'w') as f:
    df.to_csv(f, index=False, header=False)

con = sqlalchemy.create_engine('postgresql://username:[email protected]:5439/yourdatabase')
# make sure the schema for mytable exists

# if you need to delete the table but not the schema leave DELETE mytable
# if you want to only append, I think just removing the DELETE mytable would work

con.execute("""
    DELETE mytable;
    COPY mytable
    from 's3://%s'
    iam_role 'arn:aws:iam::xxxx:role/role_name'
    csv;""" % filename)

роль должна разрешать доступ Redshift к S3, см. здесь для более подробной информации

Я обнаружил, что для файла размером 300 КБ (фрейм данных 12000x2) это занимает 4 секунды по сравнению с 8 минутами, которые я получал с помощью функции pandas to_sql().

Для целей этого разговора Postgres = RedShift У вас есть два варианта:

Вариант 1:

От Pandas: http://pandas.pydata.org/pandas-docs/stable/io.html#io-sql

Модуль pandas.io.sql предоставляет набор оболочек запросов, чтобы упростить извлечение данных и уменьшить зависимость от API, специфичного для БД. Абстракция базы данных предоставляется SQLAlchemy, если она установлена. Кроме того, вам понадобится библиотека драйверов для вашей базы данных. Примерами таких драйверов являются psycopg2 для PostgreSQL или pymysql для MySQL.

Написание фреймов данных

Предполагая, что следующие данные находятся в данных DataFrame, мы можем вставить их в базу данных, используя to_sql().

id  Date    Col_1   Col_2   Col_3
26  2012-10-18  X   25.7    True
42  2012-10-19  Y   -12.4   False
63  2012-10-20  Z   5.73    True

In [437]: data.to_sql('data', engine)

В некоторых базах данных запись больших фреймов данных может привести к ошибкам из-за превышения ограничений размера пакета. Этого можно избежать, установив параметр chunksize при вызове to_sql. Например, следующий код записывает данные в базу данных пакетами по 1000 строк за раз:

In [438]: data.to_sql('data_chunked', engine, chunksize=1000)

Вариант 2

Или вы можете просто сделать свой собственный. Если у вас есть фрейм данных, называемый данными, просто переберите его, используя iterrows:

for row in data.iterrows():

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

http://initd.org/psycopg/docs/usage.html#using-copy-to-and-copy-from