Scrivere DataFrame nelle tabelle del database
Salvi un DataFrame ripulito in una tabella nuova o esistente con DataFrame.to_sql(), controllando i parametri if_exists e chunksize.
Scrivere DataFrame nelle tabelle del database è una lezione Pandas & NumPy Academy gratuita su CoddyKit. Questa è la lezione 3 di 4. Puoi leggere la lezione completa qui gratuitamente — poi esercitati direttamente nel browser con un editor di codice integrato e un tutor IA disponibile 24/7. Fa parte del percorso di apprendimento Pandas & NumPy Academy, e i tuoi progressi si sincronizzano tra il web e l'app CoddyKit. Il corso Pandas & NumPy Academy include 4 lezioni in totale.
Perché scrivere i DataFrame nei database?
Dopo aver pulito e trasformato i dati in Pandas, spesso è necessario rendere persistenti i risultati in un database: per renderli disponibili ad altre applicazioni, dashboard o membri del team; per archiviare i risultati di analisi incrementali; oppure per creare un data mart a partire da un data lake grezzo. DataFrame.to_sql() è il metodo standard di Pandas per scrivere dati in qualsiasi database supportato da SQLAlchemy con una sola chiamata.
Utilizzo di base di to_sql()
df.to_sql('table_name', con=engine, if_exists='replace', index=False) scrive il DataFrame in una tabella del database. Il parametro if_exists controlla cosa accade se la tabella esiste già: 'fail' genera un errore, 'replace' elimina e ricrea la tabella, mentre 'append' aggiunge nuove righe senza modificare quelle esistenti. Imposti sempre index=False, a meno che non desideri esplicitamente memorizzare l'indice del DataFrame come colonna nel database.
import pandas as pd
import sqlalchemy as sa
engine = sa.create_engine('sqlite:///results.db')
df = pd.DataFrame({
'date': pd.date_range('2024-01-01', periods=5),
'revenue': [1200.0, 980.5, 1450.0, 760.3, 1100.0],
'region': ['North', 'South', 'East', 'West', 'North']
})
df.to_sql('daily_revenue', con=engine,
if_exists='replace', index=False)
print('Table written successfully')Spiegazione del parametro if_exists
I tre valori di if_exists servono a casi d'uso diversi. 'replace' è adatto allo sviluppo: elimina la vecchia tabella e ne crea una nuova — le modifiche allo schema sono automatiche, ma tutti i dati precedenti vengono persi. 'append' è adatto ai caricamenti incrementali: aggiunge nuove righe alla tabella esistente senza modificarne la struttura — è utile per i processi batch giornalieri. 'fail' è una protezione: lo utilizzi per impedire che una pipeline con un bug sovrascriva accidentalmente tabelle importanti.
import pandas as pd
import sqlalchemy as sa
engine = sa.create_engine('sqlite:///results.db')
new_batch = pd.DataFrame({
'date': ['2024-06-01', '2024-06-02'],
'revenue': [1500.0, 1300.0],
'region': ['North', 'East']
})
# Append new rows without losing existing data
new_batch.to_sql('daily_revenue', con=engine,
if_exists='append', index=False)
print('Appended new rows')Controllare i tipi di dati delle colonne
Per impostazione predefinita, to_sql() associa automaticamente i dtype di Pandas ai tipi di SQLAlchemy. Talvolta i valori predefiniti non sono corretti: ad esempio, una colonna datetime64 potrebbe essere memorizzata come TEXT in SQLite. Utilizzi il parametro dtype per specificare i tipi SQL esatti usando gli oggetti tipo di SQLAlchemy. In questo modo garantisce una memorizzazione corretta, un'indicizzazione appropriata e una gestione accurata dei tipi quando i dati vengono riletti. Dopo la scrittura, verifichi sempre lo schema con un rapido PRAGMA table_info() o inspector.get_columns().
import pandas as pd
import sqlalchemy as sa
from sqlalchemy import types
engine = sa.create_engine('sqlite:///results.db')
df = pd.DataFrame({
'id': [1, 2, 3],
'name': ['Alice', 'Bob', 'Carol'],
'score': [0.95, 0.87, 0.91],
'created_at': pd.to_datetime(['2024-01-01', '2024-01-02', '2024-01-03'])
})
df.to_sql('users', con=engine, if_exists='replace', index=False,
dtype={'id': types.Integer(),
'score': types.Float(),
'created_at': types.DateTime()})Scrivere in blocchi con chunksize
Per i DataFrame di grandi dimensioni, to_sql() senza chunksize tenta di inserire tutte le righe in una singola istruzione, con il rischio di un timeout del database o di un errore di memoria. Specifichi chunksize=N per inserire N righe per transazione. In questo modo il database può eseguire il commit in modo incrementale e si riduce l'utilizzo massimo della memoria. Un chunksize compreso tra 10.000 e 50.000 righe offre in genere un buon compromesso tra velocità di inserimento e memoria, ma il valore ottimale dipende dal database e dalla latenza di rete.
import pandas as pd
import sqlalchemy as sa
import numpy as np
engine = sa.create_engine('sqlite:///results.db')
# Large DataFrame
df = pd.DataFrame({
'id': range(500000),
'value': np.random.randn(500000)
})
# Insert in chunks of 10,000 rows at a time
df.to_sql('large_table', con=engine,
if_exists='replace',
index=False,
chunksize=10000)
print('Written 500,000 rows')Upsert: inserimento o aggiornamento
to_sql() di Pandas non supporta nativamente l'upsert (inserimento se il record è nuovo, aggiornamento se esiste). Per implementare un upsert, utilizzi il Core di SQLAlchemy con un'istruzione INSERT OR REPLACE (SQLite) o ON CONFLICT DO UPDATE (PostgreSQL). Una soluzione comune in Pandas consiste nello scrivere in una tabella temporanea di staging con if_exists='replace', quindi eseguire SQL grezzo per unire la tabella di staging con quella di produzione e infine eliminare la tabella di staging.
import pandas as pd
import sqlalchemy as sa
engine = sa.create_engine('sqlite:///results.db')
new_data = pd.DataFrame({
'id': [1, 2, 5],
'value': [99.9, 88.8, 77.7]
})
# Write to staging table
new_data.to_sql('staging', con=engine,
if_exists='replace', index=False)
# Merge into production (SQLite syntax)
with engine.connect() as conn:
conn.execute(sa.text(
'INSERT OR REPLACE INTO production SELECT * FROM staging'
))
conn.commit()
print('Upsert complete')Verificare la scrittura
Dopo la scrittura, verifichi sempre il risultato rileggendo un conteggio riepilogativo e il numero di righe. Li confronti con il DataFrame di origine. In questo modo è possibile rilevare errori silenziosi causati da incompatibilità dei dtype (ad esempio, un NaN in una colonna intera che provoca inserimenti parziali) o da vincoli del database (ad esempio, violazioni di chiavi univoche che in alcune configurazioni fanno saltare silenziosamente alcune righe). Un rapido SELECT COUNT(*) FROM table dopo ogni chiamata a to_sql aggiunge un sovraccarico minimo e previene la perdita silenziosa di dati.
import pandas as pd
import sqlalchemy as sa
engine = sa.create_engine('sqlite:///results.db')
df = pd.DataFrame({'id': range(1000), 'value': range(1000)})
df.to_sql('my_table', con=engine, if_exists='replace', index=False)
# Verify
with engine.connect() as conn:
count = conn.execute(sa.text('SELECT COUNT(*) FROM my_table')).scalar()
print(f'Source rows: {len(df)}, DB rows: {count}')
assert count == len(df), 'Row count mismatch!'Aggiungere una chiave primaria dopo la scrittura
to_sql() scrive i dati, ma non aggiunge chiavi primarie né vincoli del database: crea una tabella semplice. Per una tabella di produzione, aggiunga il vincolo di chiave primaria separatamente usando SQL grezzo eseguito tramite SQLAlchemy. SQLite richiede di ricreare la tabella per aggiungere vincoli dopo la sua creazione, mentre PostgreSQL supporta ALTER TABLE ADD PRIMARY KEY. In alternativa, definisca in anticipo lo schema completo e utilizzi if_exists='append' per inserire i dati in una tabella esistente con una definizione corretta.
import pandas as pd
import sqlalchemy as sa
from sqlalchemy import Table, Column, Integer, Float, MetaData
engine = sa.create_engine('sqlite:///results.db')
meta = MetaData()
# Define table with primary key
my_table = Table('defined_table', meta,
Column('id', Integer, primary_key=True),
Column('value', Float)
)
meta.create_all(engine) # Create table with constraints
# Then insert data using append
df = pd.DataFrame({'id': range(5), 'value': [1.1, 2.2, 3.3, 4.4, 5.5]})
df.to_sql('defined_table', con=engine,
if_exists='append', index=False)Scritture transazionali
Per garantire la coerenza dei dati, racchiuda to_sql() in una transazione esplicita. Se un passaggio di una scrittura su più tabelle fallisce, può annullare tutte le modifiche. Senza una transazione, le scritture parziali possono lasciare il database in uno stato incoerente. Il gestore di contesto della connessione SQLAlchemy con conn.begin() consente di controllare manualmente la transazione. In alternativa, utilizzi engine.begin() per un blocco con commit automatico che esegue il rollback in caso di eccezione.
import pandas as pd
import sqlalchemy as sa
engine = sa.create_engine('sqlite:///results.db')
df_orders = pd.DataFrame({'id': [1, 2], 'amount': [100.0, 200.0]})
df_summary = pd.DataFrame({'total': [300.0], 'count': [2]})
try:
with engine.begin() as conn: # Auto-rollback on exception
df_orders.to_sql('orders_v2', con=conn,
if_exists='replace', index=False)
df_summary.to_sql('summary_v2', con=conn,
if_exists='replace', index=False)
print('Both tables written atomically')
except Exception as e:
print(f'Write failed, rolled back: {e}')Prestazioni: metodi di inserimento massivo
Per impostazione predefinita, to_sql() inserisce una riga per ogni istruzione SQL, risultando molto lento con i DataFrame di grandi dimensioni. Passi method='multi' per utilizzare una singola INSERT con più tuple di valori: in genere è da 10 a 100 volte più veloce. Per PostgreSQL, passi una funzione method personalizzata che utilizzi il protocollo COPY (tramite copy_expert di psycopg2) per ottenere il caricamento massivo più veloce possibile. Il metodo ottimale dipende dalla versione del database e dalla configurazione di rete.
import pandas as pd
import sqlalchemy as sa
import numpy as np
import time
engine = sa.create_engine('sqlite:///perf.db')
df = pd.DataFrame({'a': range(100000), 'b': np.random.randn(100000)})
# Default (one row per INSERT) — slow
start = time.time()
df.to_sql('test_default', con=engine, if_exists='replace', index=False)
print(f'Default: {time.time()-start:.2f}s')
# multi-row INSERT — faster
start = time.time()
df.to_sql('test_multi', con=engine, if_exists='replace',
index=False, method='multi', chunksize=1000)
print(f'Multi: {time.time()-start:.2f}s')Registrazione e controllo delle scritture
Nelle pipeline di produzione, tenga traccia di cosa è stato scritto e quando mantenendo una tabella di audit. Dopo ogni to_sql() completato correttamente, inserisca nella tabella di audit una riga con il nome della tabella, il numero di righe, il timestamp e l'ID di esecuzione della pipeline. In questo modo è facile rilevare esecuzioni mancanti, scritture duplicate o modifiche allo schema nel tempo. La tabella di audit stessa è un DataFrame Pandas scritto tramite to_sql: la stessa tecnica applicata ricorsivamente al monitoraggio operativo.
import pandas as pd
import sqlalchemy as sa
from datetime import datetime
engine = sa.create_engine('sqlite:///results.db')
def write_with_audit(df, table_name, engine, run_id):
df.to_sql(table_name, con=engine, if_exists='append', index=False)
audit = pd.DataFrame([{
'run_id': run_id,
'table_name': table_name,
'rows_written': len(df),
'written_at': datetime.utcnow().isoformat()
}])
audit.to_sql('audit_log', con=engine, if_exists='append', index=False)
print(f'Wrote {len(df)} rows to {table_name}')
df = pd.DataFrame({'id': [1, 2], 'val': [10, 20]})
write_with_audit(df, 'my_table', engine, run_id='run_001')Verifica rapida
Verifichi la comprensione dei concetti di analisi dei dati trattati in questa lezione.
Riepilogo della lezione
In questa lezione ha imparato che: df.to_sql() scrive un DataFrame in qualsiasi tabella di database connessa tramite SQLAlchemy, con il parametro if_exists che controlla il comportamento di creazione, aggiunta e sostituzione; chunksize e method='multi' migliorano le prestazioni per i DataFrame di grandi dimensioni; e le scritture transazionali con engine.begin() garantiscono aggiornamenti atomici su più tabelle, eseguendo il rollback in caso di errore. Ora confronteremo Pandas e SQL per capire quando sia preferibile ciascuno strumento.
Domande Frequenti
La lezione «Scrivere DataFrame nelle tabelle del database» è gratuita?
Sì — il testo completo di «Scrivere DataFrame nelle tabelle del database» è gratuito qui sul web. Per esercitarvi in modo interattivo (un editor di codice integrato e un tutor IA 24/7) e sbloccare il resto del corso Pandas & NumPy Academy, passa a CoddyKit PRO. Il corso Pandas & NumPy Academy include 4 lezioni in totale.
Cosa imparerò in «Scrivere DataFrame nelle tabelle del database»?
Salvi un DataFrame ripulito in una tabella nuova o esistente con DataFrame.to_sql(), controllando i parametri if_exists e chunksize. Eserciti Pandas & NumPy Academy con codice pratico che esegui direttamente nel browser, e un tutor IA 24/7 risponde alle tue domande mentre lavori sulla lezione.
Ho bisogno di esperienza per iniziare Pandas & NumPy Academy?
Non è richiesta alcuna esperienza precedente. Pandas & NumPy Academy su CoddyKit è strutturato per principianti e studenti avanzati, quindi puoi iniziare da qui o dall'inizio e procedere al tuo ritmo. Questa è la lezione 3 di 4.
Quanto tempo richiede la lezione «Scrivere DataFrame nelle tabelle del database»?
La maggior parte delle lezioni CoddyKit richiede circa 5–10 minuti. Ogni lezione è breve e interattiva, quindi fai progressi costanti e riprendi esattamente da dove hai lasciato su web e app.
Posso scrivere ed eseguire codice in questa lezione Pandas & NumPy Academy?
Sì. Ogni lezione Pandas & NumPy Academy include un editor di codice integrato, quindi scrivi ed esegui codice reale direttamente nel tuo browser e ricevi feedback istantaneo dall'IA — nessuna configurazione locale necessaria.
Tutte le lezioni di questo corso
- Connettersi a un database con SQLAlchemy
- Eseguire query SQL da Pandas
- Scrivere DataFrame nelle tabelle del database
- Pandas o SQL: scegliere lo strumento giusto