Estoy intentando consultar un subconjunto de una tabla de base de datos MySql, introducir los resultados en un Pandas DataFrame, modificar algunos datos y luego volver a escribir las filas actualizadas en la misma tabla. El tamaño de mi tabla es de ~1 mm de filas, y la cantidad de filas que alteraré será relativamente pequeña (<50 000), por lo que recuperar la tabla completa y realizar un df.to_sql(tablename,engine, if_exists='replace') no es t una opción viable. ¿Existe una forma sencilla de ACTUALIZAR las filas que se han modificado sin iterar sobre cada fila en el DataFrame?
Soy consciente de este proyecto, que intenta simular un flujo de trabajo "upsert", pero parece que solo logra la tarea de insertar nuevas filas no duplicadas en lugar de actualizar partes de las filas existentes:
Aquí hay un esqueleto de lo que estoy tratando de lograr en una escala mucho mayor:
import pandas as pd from sqlalchemy import create_engine import threading #Get sample data d = {'A' : [1, 2, 3, 4], 'B' : [4, 3, 2, 1]} df = pd.DataFrame(d) engine = create_engine(SQLALCHEMY_DATABASE_URI) #Create a table with a unique constraint on A. engine.execute("""DROP TABLE IF EXISTS test_upsert """) engine.execute("""CREATE TABLE test_upsert ( A INTEGER, B INTEGER, PRIMARY KEY (A)) """) #Insert data using pandas.to_sql df.to_sql('test_upsert', engine, if_exists='append', index=False) #Alter row where 'A' == 2 df_in_db.loc[df_in_db['A'] == 2, 'B'] = 6 Ahora me gustaría volver a escribir df_in_db en mi tabla 'test_upsert' con los datos actualizados reflejados.
Esta pregunta SO es muy similar, y uno de los comentarios recomienda usar una "clase de tabla sqlalchemy" para realizar la tarea.
Actualizar tabla usando la clase de tabla sqlalchemy
¿Alguien puede ampliar cómo implementaría esto para mi caso específico anterior si esa es la mejor (¿única?) Manera de implementarlo?
Creo que la forma más fácil sería:
primero borre aquellas filas que van a ser "alteradas". Esto se puede hacer en un ciclo, pero no es muy eficiente para conjuntos de datos más grandes (más de 5K filas), por lo que guardaría esta porción del DF en una tabla MySQL temporal:
# assuming we have already changed values in the rows and saved those changed rows in a separate DF: `x` x = df[mask] # `mask` should help us to find changed rows... # make sure `x` DF has a Primary Key column as index x = x.set_index('a') # dump a slice with changed rows to temporary MySQL table x.to_sql('my_tmp', engine, if_exists='replace', index=True) conn = engine.connect() trans = conn.begin() try: # delete those rows that we are going to "upsert" engine.execute('delete from test_upsert where a in (select a from my_tmp)') trans.commit() # insert changed rows x.to_sql('test_upsert', engine, if_exists='append', index=True) except: trans.rollback() raisePD: no probé este código, por lo que podría tener algunos errores pequeños, pero debería darte una idea...
Una solución específica de MySQL que utiliza el "método" to_sql arg de Panda y las funciones mysql insert on_duplicate_key_update de sqlalchemy:
def create_method(meta): def method(table, conn, keys, data_iter): sql_table = db.Table(table.name, meta, autoload=True) insert_stmt = db.dialects.mysql.insert(sql_table).values([dict(zip(keys, data)) for data in data_iter]) upsert_stmt = insert_stmt.on_duplicate_key_update({x.name: x for x in insert_stmt.inserted}) conn.execute(upsert_stmt) return method engine = db.create_engine(...) conn = engine.connect() with conn.begin(): meta = db.MetaData(conn) method = create_method(meta) df.to_sql(table_name, conn, if_exists='append', method=method)Estaba luchando con esto antes y ahora he encontrado una manera.
Básicamente, cree un marco de datos separado en el que guarde los datos que solo tiene que actualizar.
df #actualización de datos en marco de datos
s_update = "" #Cadena de actualizaciones
Bucle a través del marco de datos.
for i in range(len(df)): s_update += "update your_table_name set column_name = '%s' where column_name = '%s';"%(df[col_name1][i], df[col_name2][i])Ahora pase s_update a cursor.execute o engine.execute (dondequiera que ejecute la consulta SQL)
Esto actualizará sus datos al instante.