如何使用Python將MySQL表數據遷移到MongoDB集合

作者：Shameel Ahmed 2021-07-09 18:26:41

本文詳細的講解了如何在需要的地方獲取MySQL 數據。

【51CTO.com快譯】介紹

MySQL 是一個 RDBMS 平臺，它以規范化的方式以表格格式存儲數據，而 MongoDB 是一個 NoSQL 數據庫，它以無模式的方式將信息存儲為按集合分組的文檔。數據的表示方式完全不同，因此將 MySQL 表數據遷移到 MongoDB 集合聽起來可能是一項艱巨的任務。但是，Python 憑借其強大的連接性和數據處理能力讓這一切變得輕而易舉。

在本文中，將詳細的講解如何使用簡單的 Python 腳本將 MySQL 表數據遷移到 MongoDB 集合所需的步驟。這些腳本是在 Windows 上使用 Python 3.9.5 開發的。但是，它應該適用于任何平臺上的任何 Python 3+ 版本一起使用。

步驟 1：安裝所需的模塊

第一步是安裝連接MySQL 和 MongoDB 數據庫實例所需的模塊。我們將使用mysql.connector連接到 MySQL 數據庫。對于 MongoDB，使用pymongo，這是從 Python 連接到 MongoDB 的推薦模塊。

如果所需模塊尚未安裝，請運行以下PIP命令來安裝它們。

pip 安裝 mysql 連接器 pip 安裝 pymongo

PIP 是 Python 包或模塊的包管理器。

步驟2：從MySQL表中讀取數據

第一步是從源 MySQL 表中讀取數據，并以可用于將數據加載到目標 MongoDB 數據庫中的格式進行準備。MongoDB 是一個 NoSQL 數據庫，它將數據存儲為 JSON 文檔，因此最好以 JSON 格式生成源數據。值得一提的是，Python 具有強大的數據處理能力，可以輕松地將數據轉換為 JSON 格式。

import mysql.connector  
mysqldb = mysql.connector.connect(    host="localhost",    database="employees",    user="root",    password="" ) 
mycursor = mysqldb.cursor(dictionary=True) mycursor.execute("SELECT * from categories;") myresult = mycursor.fetchall()  
print(myresult)

當腳本在沒有任何錯誤的情況下完成時，輸出的結果如下：

[ 
   { 
      "id":4, 
      "name":"Medicine", 
      "description":"<p>Medicine<br></p>", 
      "created_at":"", 
      "updated_at":"" 
   }, 
   { 
      "id":6, 
      "name":"Food", 
      "description":"<p>Food</p>", 
      "created_at":"", 
      "updated_at":"" 
   }, 
   { 
      "id":8, 
      "name":"Groceries", 
      "description":"<p>Groceries<br></p>", 
      "created_at":"", 
      "updated_at":"" 
   }, 
   { 
      "id":9, 
      "name":"Cakes & Bakes", 
      "description":"<p>Cakes & Bakes<br></p>", 
      "created_at":d"", 
      "updated_at":"" 
   } 
]

請注意，輸出的是一個 JSON 數組，因為我們將dictionary=True參數傳遞給了游標。否則，結果將采用列表格式。現在有了 JSON 格式的源數據，就可以遷移到 MongoDB 集合。

步驟3：寫入 MongoDB 集合

獲得 JSON 格式的源數據后，下一步是將數據插入到 MongoDB 集合中。集合是一組文檔，相當于 RDBMS 中表（或關系）。我可以通過調用insert_many()集合類的方法來實現，該方法返回插入文檔的對象 ID 列表。請注意，當作為參數傳遞空列表時，此方法將引發異常，因此在方法調用之前進行長度檢查。

import pymongo 
 
mongodb_host = "mongodb://localhost:27017/" 
mongodb_dbname = "mymongodb" 
 
myclient = pymongo.MongoClient(mongodb_host) 
mydb = myclient[mongodb_dbname] 
mycol = mydb["categories"] 
 
if len(myresult) > 0: 
        x = mycol.insert_many(myresult) #myresult comes from mysql cursor 
        print(len(x.inserted_ids))

完成此步驟后，檢查一下 MongoDB 實例，以驗證數據庫和集合是否已創建，文檔是否已插入。注意MongoDB 是無模式的，這就意味著不必定義模式來插入文檔，模式是動態推斷并自動創建的。MongoDB 還可以創建代碼中引用的數據庫和集合（如果它們還不存在的話）。

步驟4：把數據放在一起

下面是從 MySQL 中讀取表并將其插入到 MongoDB 中集合的完整腳本。

import mysql.connector 
import pymongo 
 
delete_existing_documents = True 
mysql_host="localhost" 
mysql_database="mydatabase" 
mysql_schema = "myschema" 
mysql_user="myuser" 
mysql_password="********" 
 
mongodb_host = "mongodb://localhost:27017/" 
mongodb_dbname = "mymongodb" 
 
mysqldb = mysql.connector.connect( 
    host=mysql_host, 
    database=mysql_database, 
    user=mysql_user, 
    password=mysql_password 
) 
 
mycursor = mysqldb.cursor(dictionary=True) 
mycursor.execute("SELECT * from categories;") 
myresult = mycursor.fetchall() 
 
myclient = pymongo.MongoClient(mongodb_host) 
mydb = myclient[mongodb_dbname] 
mycol = mydb["categories"] 
 
if len(myresult) > 0: 
        x = mycol.insert_many(myresult) #myresult comes from mysql cursor 
        print(len(x.inserted_ids))

步驟 5：增強腳本以加載 MySQL 架構中的所有表

該腳本從 MySQL 中讀取一個表，并將結果加載到 MongoDB 集合中。然后，下一步是遍歷源數據庫中所有表的列表，并將結果加載到新的 MySQL 集合中。我們可以通過查詢information_schema.tables元數據表來實現這一點，該表提供給定模式中的表列表。然后可以遍歷結果并調用上面的腳本來遷移每個表的數據。

#Iterate through the list of tables in the schema 
table_list_cursor = mysqldb.cursor() 
table_list_cursor.execute("SELECT table_name FROM information_schema.tables WHERE table_schema = %s ORDER BY table_name;", (mysql_schema,)) 
tables = table_list_cursor.fetchall() 
 
for table in tables: 
    #Execute the migration script for 'table'

也可以通過將遷移邏輯抽象為一個函數來實現這一點。

#Function migrate_table  
def migrate_table(db, col_name): 
    mycursor = db.cursor(dictionary=True) 
    mycursor.execute("SELECT * FROM " + col_name + ";") 
    myresult = mycursor.fetchall() 
 
    mycol = mydb[col_name] 
     
    if delete_existing_documents: 
        #delete all documents in the collection 
        mycol.delete_many({}) 
 
    #insert the documents 
    if len(myresult) > 0: 
        x = mycol.insert_many(myresult) 
        return len(x.inserted_ids) 
    else: 
        return 0

步驟6：輸出腳本進度并使其可讀

腳本的進度是通過使用print 語句來傳達的。通過顏色編碼使輸出易于閱讀。例如，以將成功語句打印為綠色，將失敗語句打印為紅色。

class bcolors: 
    HEADER = '\033[95m' 
    OKBLUE = '\033[94m' 
    OKCYAN = '\033[96m' 
    OKGREEN = '\033[92m' 
    WARNING = '\033[93m' 
    FAIL = '\033[91m' 
    ENDC = '\033[0m' 
    BOLD = '\033[1m' 
    UNDERLINE = '\033[4m' 
 
print(f"{bcolors.HEADER}This is a header{bcolors.ENDC}") 
print(f"{bcolors.OKBLUE}This prints in blue{bcolors.ENDC}") 
print(f"{bcolors.OKGREEN}This message is green{bcolors.ENDC}")

最終結果

源 MySQL 數據庫
遷移后的目標 MongoDB 數據庫
在VSCode 中的 Python 腳本和輸出

完整的腳本

import mysql.connector 
import pymongo 
import datetime 
 
class bcolors: 
    HEADER = '\033[95m' 
    OKBLUE = '\033[94m' 
    OKCYAN = '\033[96m' 
    OKGREEN = '\033[92m' 
    WARNING = '\033[93m' 
    FAIL = '\033[91m' 
    ENDC = '\033[0m' 
    BOLD = '\033[1m' 
    UNDERLINE = '\033[4m' 
 
begin_time = datetime.datetime.now() 
print(f"{bcolors.HEADER}Script started at: {begin_time} {bcolors.ENDC}") 
 
delete_existing_documents = True; 
mysql_host="localhost" 
mysql_database="mydatabase" 
mysql_schema = "myschhema" 
mysql_user="root" 
mysql_password="" 
 
mongodb_host = "mongodb://localhost:27017/" 
mongodb_dbname = "mymongodb" 
 
print(f"{bcolors.HEADER}Initializing database connections...{bcolors.ENDC}") 
print("") 
 
#MySQL connection 
print(f"{bcolors.HEADER}Connecting to MySQL server...{bcolors.ENDC}") 
mysqldb = mysql.connector.connect( 
    host=mysql_host, 
    database=mysql_database, 
    user=mysql_user, 
    password=mysql_password 
) 
print(f"{bcolors.HEADER}Connection to MySQL Server succeeded.{bcolors.ENDC}") 
 
#MongoDB connection 
print(f"{bcolors.HEADER}Connecting to MongoDB server...{bcolors.ENDC}") 
myclient = pymongo.MongoClient(mongodb_host) 
mydb = myclient[mongodb_dbname] 
print(f"{bcolors.HEADER}Connection to MongoDB Server succeeded.{bcolors.ENDC}") 
 
print(f"{bcolors.HEADER}Database connections initialized successfully.{bcolors.ENDC}") 
 
#Start migration 
print(f"{bcolors.HEADER}Migration started...{bcolors.ENDC}") 
dblist = myclient.list_database_names() 
if mongodb_dbname in dblist: 
    print(f"{bcolors.OKBLUE}The database exists.{bcolors.ENDC}") 
else: 
    print(f"{bcolors.WARNING}The database does not exist, it is being created.{bcolors.ENDC}") 
 
#Function migrate_table  
def migrate_table(db, col_name): 
    mycursor = db.cursor(dictionary=True) 
    mycursor.execute("SELECT * FROM " + col_name + ";") 
    myresult = mycursor.fetchall() 
 
    mycol = mydb[col_name] 
     
    if delete_existing_documents: 
        #delete all documents in the collection 
        mycol.delete_many({}) 
 
    #insert the documents 
    if len(myresult) > 0: 
        x = mycol.insert_many(myresult) 
        return len(x.inserted_ids) 
    else: 
        return 0 
 
#Iterate through the list of tables in the schema 
table_list_cursor = mysqldb.cursor() 
table_list_cursor.execute("SELECT table_name FROM information_schema.tables WHERE table_schema = %s ORDER BY table_name LIMIT 15;", (mysql_schema,)) 
tables = table_list_cursor.fetchall() 
 
total_count = len(tables) 
success_count = 0 
fail_count = 0 
 
for table in tables: 
    try: 
        print(f"{bcolors.OKCYAN}Processing table: {table[0]}...{bcolors.ENDC}") 
        inserted_count = migrate_table(mysqldb, table[0]) 
        print(f"{bcolors.OKGREEN}Processing table: {table[0]} completed. {inserted_count} documents inserted.{bcolors.ENDC}") 
        success_count += 1 
    except Exception as e: 
        print(f"{bcolors.FAIL} {e} {bcolors.ENDC}") 
        fail_count += 1 
         
print("") 
print("Migration completed.") 
print(f"{bcolors.OKGREEN}{success_count} of {total_count} tables migrated successfully.{bcolors.ENDC}") 
if fail_count > 0: 
    print(f"{bcolors.FAIL}Migration of {fail_count} tables failed. See errors above.{bcolors.ENDC}") 
 
end_time = datetime.datetime.now() 
print(f"{bcolors.HEADER}Script completed at: {end_time} {bcolors.ENDC}") 
print(f"{bcolors.HEADER}Total execution time: {end_time-begin_time} {bcolors.ENDC}")

警告

該腳本適用于中小型 MySQL 數據庫，它有幾百個表，每個表有幾千行。對于具有數百萬行的大型數據庫，性能可能會受到影響。在開始實際遷移之前，請在表列表查詢和實際表select查詢上使用 LIMIT 關鍵字對有限行進行檢測。

下載

帶點擊此處從 GitHub 下載整個腳本：
https://github.com/zshameel/MySQL2MongoDB

【51CTO譯稿，合作站點轉載請注明原文譯者和出處為51CTO.com】

責任編輯：梁菲來源： DZone