Accélérer les pandas.DataFrame.to_sql avec fast_executemany de pyODBC

MODIFIER (08/03/2019): Gord Thompson a commenté ci-dessous, avec de bonnes nouvelles de la mise à jour de journaux de sqlalchemy: Depuis SQLAlchemy 1.3.0, publié 2019-03-04, sqlalchemy prend désormais en charge engine = create_engine(sqlalchemy_url, fast_executemany=True) pour la mssql+pyodbc dialecte. I. e., il n'est plus nécessaire de définir une fonction et l'utilisation @event.listens_for(engine, 'before_cursor_execute') Signification ci-dessous la fonction peut être supprimée et seule l'indicateur doit être défini dans le create_engine de l'instruction et de toujours conserver le speed-up.

Original Post:

Viens de faire un compte pour poster ceci. Je voulais commentaire sous le fil au-dessus car il s'agit d'un suivi sur le déjà fourni la réponse. La solution ci-dessus a fonctionné pour moi avec la Version 17 SQL pilote Microsoft SQL stockage de l'écriture à partir d'un Ubuntu installation.

Le code complet que j'ai utilisé pour accélérer les choses de manière significative (parler >100x speed-up) est ci-dessous. Ceci est un extrait de à la condition que vous modifiez la chaîne de connexion avec vos détails pertinents. L'affiche ci-dessus, je vous remercie très beaucoup pour la solution que je cherchais tout à fait un certain temps déjà.

import pandas as pd
import numpy as np
import time
from sqlalchemy import create_engine, event
from urllib.parse import quote_plus


conn =  "DRIVER={ODBC Driver 17 for SQL Server};SERVER=IP_ADDRESS;DATABASE=DataLake;UID=USER;PWD=PASS"
quoted = quote_plus(conn)
new_con = 'mssql+pyodbc:///?odbc_connect={}'.format(quoted)
engine = create_engine(new_con)


@event.listens_for(engine, 'before_cursor_execute')
def receive_before_cursor_execute(conn, cursor, statement, params, context, executemany):
    print("FUNC call")
    if executemany:
        cursor.fast_executemany = True


table_name = 'fast_executemany_test'
df = pd.DataFrame(np.random.random((10**4, 100)))


s = time.time()
df.to_sql(table_name, engine, if_exists = 'replace', chunksize = None)
print(time.time() - s)

Sur la base des commentaires ci-dessous, je voulais prendre un peu de temps pour expliquer certaines limitations sur les pandas to_sql mise en œuvre et la façon dont la requête est traitée. Il y a 2 choses qui peuvent causer la MemoryError soulevé autant que je sache:

1) en Supposant que vous écrivez à une instance distante de SQL de stockage. Lorsque vous essayez d'écrire un grand pandas DataFrame avec le to_sql méthode il convertit l'intégralité du dataframe dans une liste de valeurs. Cette transformation prend beaucoup plus de RAM que l'original DataFrame n' (sur le dessus de cela, comme l'ancien DataFrame reste encore présent dans la mémoire RAM). Cette liste est fournie à la dernière executemany appel à votre connecteur ODBC. Je pense que le connecteur ODBC a quelques problèmes pour le traitement de ces requêtes importantes. Une façon de résoudre ce problème est de fournir le to_sql méthode un chunksize argument (10**5 semble être autour optimale donnant sur 600 mbit/s (!) des vitesses d'écriture sur un 2 UC 7 GO de ram MSSQL application de Stockage d'Azur ne peux pas le recommander Azure btw). Donc, la première limitation, soit la taille de requête peut être contournée par la fourniture d'un chunksize argument. Cependant, ce ne sera pas vous permettre d'écrire un dataframe la taille de 10**7 ou plus, (au moins pas sur la VM, je travaille avec ce qui a ~55GB RAM), étant à la question n ° 2.

Cela peut être contourné par la rupture de la DataFrame avec np.split (10**6 taille DataFrame morceaux) ils peuvent être écrits à l'écart de manière itérative. Je vais essayer de faire une pull request, quand j'ai une solution toute prête pour la to_sql méthode dans la base de pandas lui-même de sorte que vous n'aurez pas à faire cette pré-rupture de tous les temps. De toute façon j'ai fini par écrire une fonction similaire (pas de clé) pour les opérations suivantes:

import pandas as pd
import numpy as np

def write_df_to_sql(df, **kwargs):
    chunks = np.split(df, df.shape()[0] / 10**6)
    for chunk in chunks:
        chunk.to_sql(**kwargs)
    return True

Un exemple plus complet de l'extrait ci-dessus peuvent être consultées ici: https://gitlab.com/timelord/timelord/blob/master/timelord/utils/connector.py

C'est une classe que j'ai écrit qui intègre le patch et facilite certaines des frais généraux qui vient avec l'établissement de connexions avec SQL. Encore écrire de la documentation. J'ai également la planification sur le patch pour les pandas lui-même, mais n'ai pas trouvé une belle façon sur la façon de le faire.

J'espère que cette aide.

Je voulais juste poster cet exemple comme un autre, hautes performances et pour ceux qui peuvent utiliser la nouvelle turbodbc bibliothèque: http://turbodbc.readthedocs.io/en/latest/

Il n'y a clairement beaucoup d'options dans le flux entre les pandas .to_sql(), le déclenchement de fast_executemany par le biais de sqlalchemy, à l'aide de pyodbc directement avec les tuples/lists/etc., ou même essayer de TÉLÉCHARGEMENT en VRAC avec des fichiers plats.

Espérons-le, les éléments suivants pourraient rendre la vie un peu plus agréable que la fonctionnalité évolue dans le courant de pandas projet ou comprend quelque chose comme turbodbc intégration dans l'avenir.

import pandas as pd
import numpy as np
from turbodbc import connect, make_options
from io import StringIO

test_data = '''id,transaction_dt,units,measures
               1,2018-01-01,4,30.5
               1,2018-01-03,4,26.3
               2,2018-01-01,3,12.7
               2,2018-01-03,3,8.8'''

df_test = pd.read_csv(StringIO(test_data), sep=',')
df_test['transaction_dt'] = pd.to_datetime(df_test['transaction_dt'])

options = make_options(parameter_sets_to_buffer=1000)
conn = connect(driver='{SQL Server}', server='server_nm', database='db_nm', turbodbc_options=options)

test_query = '''DROP TABLE IF EXISTS [db_name].[schema].[test]

                CREATE TABLE [db_name].[schema].[test]
                (
                    id int NULL,
                    transaction_dt datetime NULL,
                    units int NULL,
                    measures float NULL
                )

                INSERT INTO [db_name].[schema].[test] (id,transaction_dt,units,measures)
                VALUES (?,?,?,?) '''

cursor.executemanycolumns(test_query, [df_test['id'].values, df_test['transaction_dt'].values, df_test['units'].values, df_test['measures'].values]

turbodbc devrait être TRÈS rapide dans de nombreux cas d'utilisation (en particulier avec des tableaux numpy). Veuillez observer comment simple c'est de passer les sous-jacents des tableaux numpy du dataframe colonnes en tant que paramètres à la requête directement. Je crois aussi que cela permet d'éviter la création d'objets intermédiaires que les pointes de consommation mémoire excessive. Espérons que cela est utile!

Il semble que les Pandas 0.23.0 et 0.24.0 l'utilisation de plusieurs valeurs inserts avec PyODBC, ce qui empêche rapide executemany de l'aide – une seule INSERT ... VALUES ... déclaration est émise par morceau. Multi valeurs insérer des segments sont une amélioration par rapport à l'ancien lent executemany par défaut, mais au moins dans les tests simples de la rapide executemany méthode prévaut encore, pour ne pas mentionner aucun besoin de manuel chunksize calculs, comme cela est requis, avec de multiples valeurs des inserts. Forçant l'ancien comportement peut être fait par monkeypatching, si aucune option de configuration est fourni dans l'avenir:

import pandas.io.sql

def insert_statement(self, data, conn):
    return self.table.insert(), data

pandas.io.sql.SQLTable.insert_statement = insert_statement

---

L'avenir est ici, et au moins dans la master direction générale de la méthode d'insertion peut être contrôlé à l'aide de l'argument mot-clé method= de to_sql(). La valeur par défaut est None, ce qui oblige les executemany méthode. En passant method='multi' résultats dans l'utilisation de la multi valeurs à insérer. Il peut même être utilisé pour mettre en œuvre des SGBD approches spécifiques, tels que Postgresql COPY.

Comme l'a souligné @Pylander

Turbodbc est le meilleur choix pour l'ingestion de données, et de loin!

Je suis tellement excité à ce sujet que j'ai écrit un blog sur mon github et support:
veuillez vérifier https://medium.com/@erickfis/etl-process-with-turbodbc-1d19ed71510e

pour un exemple de travail et de comparaison avec les pandas.to_sql

Longue histoire courte,

avec turbodbc
J'ai 10000 lignes (77 colonnes) en 3 secondes

avec les pandas.to_sql
J'ai eu la même 10000 lignes (77 colonnes) de 198 secondes...

Et voici ce que je fais en détail

Les importations:

import sqlalchemy
import pandas as pd
import numpy as np
import turbodbc
import time

Charge et traiter certaines données Substitut de mon échantillon.pkl pour le vôtre:

df = pd.read_pickle('sample.pkl')

df.columns = df.columns.str.strip()  # remove white spaces around column names
df = df.applymap(str.strip) # remove white spaces around values
df = df.replace('', np.nan)  # map nans, to drop NAs rows and columns later
df = df.dropna(how='all', axis=0)  # remove rows containing only NAs
df = df.dropna(how='all', axis=1)  # remove columns containing only NAs
df = df.replace(np.nan, 'NA')  # turbodbc hates null values...

Créer la table à l'aide de sqlAlchemy

Malheureusement, turbodbc nécessite beaucoup de frais avec beaucoup de sql travail manuel, pour la création des tables et pour l'insertion de données.

Heureusement, Python est un pur bonheur et nous pouvons automatiser ce processus d'écriture de code sql.

La première étape est la création de la table qui recevra nos données. Toutefois, la création de la table manuellement l'écriture de code sql peut être problématique si votre table a plus que quelques colonnes. Dans mon cas, très souvent, les tables ont 240 colonnes!

C'est là que sqlAlchemy et les pandas peuvent encore nous aider: les pandas est mauvais pour la rédaction d'un grand nombre de lignes (10000 dans cet exemple), mais qu'en seulement 6 lignes, la tête de la table? De cette façon, nous automatiser le processus de création des tables.

Créer sqlAlchemy connexion:

mydb = 'someDB'

def make_con(db):
    """Connect to a specified db."""
    database_connection = sqlalchemy.create_engine(
        'mssql+pymssql://{0}:{1}@{2}/{3}'.format(
            myuser, mypassword,
            myhost, db
            )
        )
    return database_connection

pd_connection = make_con(mydb)

Créer une table dans SQL Server

À l'aide de pandas + sqlAlchemy, mais juste pour la préparation de la chambre pour turbodbc, comme mentionné précédemment. Veuillez noter que les df.la tête() ici: nous sommes à l'aide de pandas + sqlAlchemy pour l'insertion de seulement 6 lignes de nos données. Cela permettra de courir assez vite et est fait pour automatiser la création de la table.

table = 'testing'
df.head().to_sql(table, con=pd_connection, index=False)

Maintenant que la table est déjà en place, nous allons prendre au sérieux ici.

Turbodbc connexion:

def turbo_conn(mydb):
    """Connect to a specified db - turbo."""
    database_connection = turbodbc.connect(
                                            driver='ODBC Driver 17 for SQL Server',
                                            server=myhost,
                                            database=mydb,
                                            uid=myuser,
                                            pwd=mypassword
                                        )
    return database_connection

La préparation de sql comands et de données pour turbodbc. Nous allons automatiser la création de code, être créatif:

def turbo_write(mydb, df, table):
    """Use turbodbc to insert data into sql."""
    start = time.time()
    # preparing columns
    colunas = '('
    colunas += ', '.join(df.columns)
    colunas += ')'

    # preparing value place holders
    val_place_holder = ['?' for col in df.columns]
    sql_val = '('
    sql_val += ', '.join(val_place_holder)
    sql_val += ')'

    # writing sql query for turbodbc
    sql = f"""
    INSERT INTO {mydb}.dbo.{table} {colunas}
    VALUES {sql_val}
    """

    # writing array of values for turbodbc
    valores_df = [df[col].values for col in df.columns]

    # cleans the previous head insert
    with connection.cursor() as cursor:
        cursor.execute(f"delete from {mydb}.dbo.{table}")
        connection.commit()

    # inserts data, for real
    with connection.cursor() as cursor:
        try:
            cursor.executemanycolumns(sql, valores_df)
            connection.commit()
        except Exception:
            connection.rollback()
            print('something went wrong')

    stop = time.time() - start
    return print(f'finished in {stop} seconds')

De l'écriture de données à l'aide de turbodbc - j'ai 10000 lignes (77 colonnes) dans les 3 secondes:

turbo_write(mydb, df.sample(10000), table)

Pandas méthode de la comparaison, j'ai eu la même 10000 lignes (77 colonnes) de 198 secondes...

table = 'pd_testing'

def pandas_comparisson(df, table):
    """Load data using pandas."""
    start = time.time()
    df.to_sql(table, con=pd_connection, index=False)
    stop = time.time() - start
    return print(f'finished in {stop} seconds')

pandas_comparisson(df.sample(10000), table)

De l'environnement et des conditions de

Python 3.6.7 :: Anaconda, Inc.
TURBODBC version ‘3.0.0’
sqlAlchemy version ‘1.2.12’
pandas version ‘0.23.4’
Microsoft SQL Server 2014
user with bulk operations privileges

Veuillez vérifier https://erickfis.github.io/loose-code/ pour les mises à jour dans ce code!