python - SQL Alchemy - 从 Oracle 迁移到 MySQL 的 Python 脚本
问题描述
我正在尝试使用 cx_Oracle 和 SQL Alchemy 执行从 Oracle 到 MySQL 的批量提取/加载。
我在网上找到了这个示例,它适用于大多数数据类型,但在 Blob 数据类型中失败:
https://vbaoverall.com/transfer-data-from-oracle-to-mysql-using-sqlalchemy-python/
我有大约 43 个表,其中大约 12 个具有BLOB
数据类型。
import cx_Oracle
import pandas as pd
from sqlalchemy import create_engine
import pymysql
import warnings
warnings.filterwarnings('ignore')
# list out all 43 tables:
table_list = [
"FILE",
"ATTACHMENT",
"DOCUMENTS",
"USERS",
"INFO",
"ONE",
"TWO",
"THREE",
"FOUR",
"...."
]
# Set Oralce Connection
dsn_tns = cx_Oracle.makedsn('source.example.com', '1530', service_name='test')
oracle_connection = cx_Oracle.connect(user='root', password='toot', dsn=dsn_tns)
# Open Oracle cursor
cursor = oracle_connection.cursor()
# set mysql connection with foreign key checks
mysql_engine = create_engine("mysql+pymysql://root:toot@target.example.com:3306/target")
mysql_engine.execute("SET FOREIGN_KEY_CHECKS=0")
# loop thru tables:
for table in table_list:
# select from oracle
sql = "SELECT * FROM " + table
# read into pandas df
data=pd.read_sql(sql, oracle_connection)
# insert into mysql
mysql_engine.execute("TRUNCATE TABLE "+table)
data.to_sql(table, con=mysql_engine, if_exists='append', index=False, chunksize=10000)
print("{}: sucessfully inserted {} rows.".format(table, data.shape[0]))
# update foreign key checks
mysql_engine.execute("SET FOREIGN_KEY_CHECKS=1")
#close connection
oracle_connection.close()
mysql_engine.dispose()
这是我得到的错误:
return "'%s'" % escape_string(str(value), mapping)
TypeError: __str__ returned non-string (type bytes)
解决方案
感谢@Gord Thompson,我发现我只需要指定dtype=
import cx_Oracle
import pandas as pd
from sqlalchemy import create_engine
import sqlalchemy
import pymysql
import warnings
warnings.filterwarnings('ignore')
table_list = [
"FILE",
"ATTACHMENT",
"DOCUMENTS",
"USERS",
"INFO",
"ONE",
"TWO",
"THREE",
"FOUR",
"...."
]
# Set Oralce Connection
dsn_tns = cx_Oracle.makedsn('source.example.com', '1530', service_name='test')
oracle_connection = cx_Oracle.connect(user='root', password='toot', dsn=dsn_tns)
# Open Oracle cursor
cursor = oracle_connection.cursor()
# set mysql connection with foreign key checks
mysql_engine = create_engine("mysql+pymysql://root:toot@target.example.com:3306/target")
mysql_engine.execute("SET FOREIGN_KEY_CHECKS=0")
for table in table_list:
# select from oracle
sql = "SELECT * FROM " + table
# read into pandas df
data=pd.read_sql(sql, oracle_connection)
dtype = {}
if table == "ATTACHMENT":
dtype['FILE_CONTENT'] = sqlalchemy.types.PickleType
# insert into mysql
mysql_engine.execute("TRUNCATE TABLE "+table)
data.to_sql(table, con=mysql_engine, if_exists='append', index = False, chunksize =10000, dtype=dtype)
print("{}: sucessfully inserted {} rows.".format(table, data.shape[0]))
# update foreign key checks
mysql_engine.execute("SET FOREIGN_KEY_CHECKS=1")
#close connection
oracle_connection.close()
mysql_engine.dispose()
推荐阅读
- sql - SQL 查询访问 - 显示截至 9 月底的记录
- php - Slimframework 3重定向在控制器中不起作用
- android - when i fetch image from camera that time crash my application in Redmi note 3
- prolog - 在 Prolog 中指定列表长度作为目标
- ios - Multiple errors after migrating to Xcode 10
- c# - 抛出异常:WebDriver.dll 中的“OpenQA.Selenium.WebDriverException”
- ios - Swift:如何根据 Firebase 中的用户 ID 读取特定的用户数据?
- android - Android Firebase Job Dispatcher 会通过更改日期时间触发吗?
- javascript - 在 Javascript 类中创建全局变量时出错
- android - 如何使用 fcm 在 android 中实现多个通知 onclick?