SQL Alchemy - 从 Oracle 迁移到 MySQL 的 Python 脚本答案

【问题标题】：SQL Alchemy - Python script to migrate from Oracle to MySQLSQL Alchemy - 从 Oracle 迁移到 MySQL 的 Python 脚本
【发布时间】：2021-08-26 17:09:18
【问题描述】：

我正在尝试使用 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)

【问题讨论】：

您是否尝试过使用to_sql() 的dtype= 参数来显式设置列类型？
我在尝试这个时收到了一个不同的错误：dtype = {} dtype['FILE_CONTENT'] = sqlalchemy.types.BLOB 我尝试了BLOB 和BINARY 并收到了这个错误：sqlalchemy.exc.StatementError: (builtins.TypeError) cannot convert 'cx_Oracle.LOB' object to bytes
实际上，当我将类型更改为PickleType 时，它起作用了。谢谢@GordThompson

标签： python sqlalchemy etl cx-oracle pymysql

【解决方案1】：

感谢@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()

【讨论】：