SQLAlchemyでデータをdataframeに取得し列を追加してMySQLにインポート

Python

2019.05.22

業務であるテーブルから、日付を追記して他のテーブルに入れる処理をSQLAlchemyで対応しました。

MySQL1から別のMySQL2にデータをexportしてimportする際に、MySQL1のデータに日付カラムを追加して、MySQL2にデータをimportする必要があったので、SQLAlchemyを使って解決しました。

コード

import time 
import pandas as pd 
import mysql.connector 
from sqlalchemy import create_engine 

def main(): 
  time_stamp = time.strftime('%Y-%m-%d %H:%M:%S') 
  # Import from Product Center 
  engine = create_engine('mysql+mysqlconnector://username:password@server1:3306/schema', echo = False)
  query = 'select * from product_inventory_info where date(updated_at) = \'{0}\''.format(time_stamp[:10])
  df = pd.read_sql_query(query, engine) 
  engine.dispose() 

  # Export to local DB 
  df['time_stamp'] = time_stamp 
  engine = create_engine('mysql+mysqlconnector://username:password@server2:3306/schema', echo = False) 
  df.to_sql(name='product_inventory_info_history', con=engine, if_exists='append', chunksize=5000 ) 
  engine.dispose() 
if __name__ == '__main__': 
  main()

はまったのはMySQL2にデータをimportするこの部分です。

df.to_sql(name='product_inventory_info_history', con=engine, if_exists='append', chunksize=5000 )

初めは”chunksize=5000″無しでテストをしていたのですが、どうしてもエラーが出ていました。そこで公式サイトを見て見るとchunksizeが怪しそうでした。

chunksize : int, optional

Rows will be written in batches of this size at a time. By default, all rows will be written at once.

pandas.DataFrame.to_sql — pandas 2.2.0 documentation
pandas.pydata.org