自拍偷在线精品自拍偷,亚洲欧美中文日韩v在线观看不卡

如何使用Python將MySQL表數(shù)據(jù)遷移到MongoDB集合

譯文
作者：Shameel Ahmed
云計(jì)算 MongoDB
本文詳細(xì)的講解了如何在需要的地方獲取MySQL 數(shù)據(jù)。

[[410477]]

【51CTO.com快譯】介紹

MySQL 是一個(gè) RDBMS 平臺(tái)，它以規(guī)范化的方式以表格格式存儲(chǔ)數(shù)據(jù)，而 MongoDB 是一個(gè) NoSQL 數(shù)據(jù)庫(kù)，它以無(wú)模式的方式將信息存儲(chǔ)為按集合分組的文檔。數(shù)據(jù)的表示方式完全不同，因此將 MySQL 表數(shù)據(jù)遷移到 MongoDB 集合聽(tīng)起來(lái)可能是一項(xiàng)艱巨的任務(wù)。但是，Python 憑借其強(qiáng)大的連接性和數(shù)據(jù)處理能力讓這一切變得輕而易舉。

在本文中，將詳細(xì)的講解如何使用簡(jiǎn)單的 Python 腳本將 MySQL 表數(shù)據(jù)遷移到 MongoDB 集合所需的步驟。這些腳本是在 Windows 上使用 Python 3.9.5 開(kāi)發(fā)的。但是，它應(yīng)該適用于任何平臺(tái)上的任何 Python 3+ 版本一起使用。

步驟 1：安裝所需的模塊

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

如果所需模塊尚未安裝，請(qǐng)運(yùn)行以下PIP命令來(lái)安裝它們。

  1. pip 安裝 mysql 連接器 pip 安裝 pymongo 

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

步驟2：從MySQL表中讀取數(shù)據(jù)

第一步是從源 MySQL 表中讀取數(shù)據(jù)，并以可用于將數(shù)據(jù)加載到目標(biāo) MongoDB 數(shù)據(jù)庫(kù)中的格式進(jìn)行準(zhǔn)備。MongoDB 是一個(gè) NoSQL 數(shù)據(jù)庫(kù)，它將數(shù)據(jù)存儲(chǔ)為 JSON 文檔，因此最好以 JSON 格式生成源數(shù)據(jù)。值得一提的是，Python 具有強(qiáng)大的數(shù)據(jù)處理能力，可以輕松地將數(shù)據(jù)轉(zhuǎn)換為 JSON 格式。

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

當(dāng)腳本在沒(méi)有任何錯(cuò)誤的情況下完成時(shí)，輸出的結(jié)果如下：

  1. [ 
  2.    { 
  3.       "id":4, 
  4.       "name":"Medicine", 
  5.       "description":"<p>Medicine<br></p>", 
  6.       "created_at":"", 
  7.       "updated_at":"" 
  8.    }, 
  9.    { 
  10.       "id":6, 
  11.       "name":"Food", 
  12.       "description":"<p>Food</p>", 
  13.       "created_at":"", 
  14.       "updated_at":"" 
  15.    }, 
  16.    { 
  17.       "id":8, 
  18.       "name":"Groceries", 
  19.       "description":"<p>Groceries<br></p>", 
  20.       "created_at":"", 
  21.       "updated_at":"" 
  22.    }, 
  23.    { 
  24.       "id":9, 
  25.       "name":"Cakes & Bakes", 
  26.       "description":"<p>Cakes & Bakes<br></p>", 
  27.       "created_at":d"", 
  28.       "updated_at":"" 
  29.    } 
  30. ] 

請(qǐng)注意，輸出的是一個(gè) JSON 數(shù)組，因?yàn)槲覀儗?span style="background-color: rgb(153, 153, 153);">dictionary=True參數(shù)傳遞給了游標(biāo)。否則，結(jié)果將采用列表格式?，F(xiàn)在有了 JSON 格式的源數(shù)據(jù)，就可以遷移到 MongoDB 集合。

步驟3：寫(xiě)入 MongoDB 集合

獲得 JSON 格式的源數(shù)據(jù)后，下一步是將數(shù)據(jù)插入到 MongoDB 集合中。集合是一組文檔，相當(dāng)于 RDBMS 中表（或關(guān)系）。我可以通過(guò)調(diào)用insert_many()集合類的方法來(lái)實(shí)現(xiàn)，該方法返回插入文檔的對(duì)象 ID 列表。請(qǐng)注意，當(dāng)作為參數(shù)傳遞空列表時(shí)，此方法將引發(fā)異常，因此在方法調(diào)用之前進(jìn)行長(zhǎng)度檢查。

  1. import pymongo 
  2.  
  3. mongodb_host = "mongodb://localhost:27017/" 
  4. mongodb_dbname = "mymongodb" 
  5.  
  6. myclient = pymongo.MongoClient(mongodb_host) 
  7. mydb = myclient[mongodb_dbname] 
  8. mycol = mydb["categories"] 
  9.  
  10. if len(myresult) > 0: 
  11.         x = mycol.insert_many(myresult) #myresult comes from mysql cursor 
  12.         print(len(x.inserted_ids)) 

完成此步驟后，檢查一下 MongoDB 實(shí)例，以驗(yàn)證數(shù)據(jù)庫(kù)和集合是否已創(chuàng)建，文檔是否已插入。注意MongoDB 是無(wú)模式的，這就意味著不必定義模式來(lái)插入文檔，模式是動(dòng)態(tài)推斷并自動(dòng)創(chuàng)建的。MongoDB 還可以創(chuàng)建代碼中引用的數(shù)據(jù)庫(kù)和集合（如果它們還不存在的話）。

步驟4：把數(shù)據(jù)放在一起

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

  1. import mysql.connector 
  2. import pymongo 
  3.  
  4. delete_existing_documents = True 
  5. mysql_host="localhost" 
  6. mysql_database="mydatabase" 
  7. mysql_schema = "myschema" 
  8. mysql_user="myuser" 
  9. mysql_password="********" 
  10.  
  11. mongodb_host = "mongodb://localhost:27017/" 
  12. mongodb_dbname = "mymongodb" 
  13.  
  14. mysqldb = mysql.connector.connect( 
  15.     host=mysql_host, 
  16.     database=mysql_database, 
  17.     user=mysql_user, 
  18.     password=mysql_password 
  19. ) 
  20.  
  21. mycursor = mysqldb.cursor(dictionary=True) 
  22. mycursor.execute("SELECT * from categories;") 
  23. myresult = mycursor.fetchall() 
  24.  
  25. myclient = pymongo.MongoClient(mongodb_host) 
  26. mydb = myclient[mongodb_dbname] 
  27. mycol = mydb["categories"] 
  28.  
  29. if len(myresult) > 0: 
  30.         x = mycol.insert_many(myresult) #myresult comes from mysql cursor 
  31.         print(len(x.inserted_ids)) 

步驟 5：增強(qiáng)腳本以加載 MySQL 架構(gòu)中的所有表

該腳本從 MySQL 中讀取一個(gè)表，并將結(jié)果加載到 MongoDB 集合中。然后，下一步是遍歷源數(shù)據(jù)庫(kù)中所有表的列表，并將結(jié)果加載到新的 MySQL 集合中。我們可以通過(guò)查詢information_schema.tables元數(shù)據(jù)表來(lái)實(shí)現(xiàn)這一點(diǎn)，該表提供給定模式中的表列表。然后可以遍歷結(jié)果并調(diào)用上面的腳本來(lái)遷移每個(gè)表的數(shù)據(jù)。

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

也可以通過(guò)將遷移邏輯抽象為一個(gè)函數(shù)來(lái)實(shí)現(xiàn)這一點(diǎn)。

  1. #Function migrate_table  
  2. def migrate_table(db, col_name): 
  3.     mycursor = db.cursor(dictionary=True) 
  4.     mycursor.execute("SELECT * FROM " + col_name + ";") 
  5.     myresult = mycursor.fetchall() 
  6.  
  7.     mycol = mydb[col_name] 
  8.      
  9.     if delete_existing_documents: 
  10.         #delete all documents in the collection 
  11.         mycol.delete_many({}) 
  12.  
  13.     #insert the documents 
  14.     if len(myresult) > 0: 
  15.         x = mycol.insert_many(myresult) 
  16.         return len(x.inserted_ids) 
  17.     else: 
  18.         return 0 

步驟6：輸出腳本進(jìn)度并使其可讀

腳本的進(jìn)度是通過(guò)使用print 語(yǔ)句來(lái)傳達(dá)的。通過(guò)顏色編碼使輸出易于閱讀。例如，以將成功語(yǔ)句打印為綠色，將失敗語(yǔ)句打印為紅色。

  1. class bcolors: 
  2.     HEADER = '\033[95m' 
  3.     OKBLUE = '\033[94m' 
  4.     OKCYAN = '\033[96m' 
  5.     OKGREEN = '\033[92m' 
  6.     WARNING = '\033[93m' 
  7.     FAIL = '\033[91m' 
  8.     ENDC = '\033[0m' 
  9.     BOLD = '\033[1m' 
  10.     UNDERLINE = '\033[4m' 
  11.  
  12. print(f"{bcolors.HEADER}This is a header{bcolors.ENDC}") 
  13. print(f"{bcolors.OKBLUE}This prints in blue{bcolors.ENDC}") 
  14. print(f"{bcolors.OKGREEN}This message is green{bcolors.ENDC}") 

最終結(jié)果

  • 源 MySQL 數(shù)據(jù)庫(kù)
  • 遷移后的目標(biāo) MongoDB 數(shù)據(jù)庫(kù)
  • 在VSCode 中的 Python 腳本和輸出

完整的腳本

  1. import mysql.connector 
  2. import pymongo 
  3. import datetime 
  4.  
  5. class bcolors: 
  6.     HEADER = '\033[95m' 
  7.     OKBLUE = '\033[94m' 
  8.     OKCYAN = '\033[96m' 
  9.     OKGREEN = '\033[92m' 
  10.     WARNING = '\033[93m' 
  11.     FAIL = '\033[91m' 
  12.     ENDC = '\033[0m' 
  13.     BOLD = '\033[1m' 
  14.     UNDERLINE = '\033[4m' 
  15.  
  16. begin_time = datetime.datetime.now() 
  17. print(f"{bcolors.HEADER}Script started at: {begin_time} {bcolors.ENDC}") 
  18.  
  19. delete_existing_documents = True; 
  20. mysql_host="localhost" 
  21. mysql_database="mydatabase" 
  22. mysql_schema = "myschhema" 
  23. mysql_user="root" 
  24. mysql_password="" 
  25.  
  26. mongodb_host = "mongodb://localhost:27017/" 
  27. mongodb_dbname = "mymongodb" 
  28.  
  29. print(f"{bcolors.HEADER}Initializing database connections...{bcolors.ENDC}") 
  30. print("") 
  31.  
  32. #MySQL connection 
  33. print(f"{bcolors.HEADER}Connecting to MySQL server...{bcolors.ENDC}") 
  34. mysqldb = mysql.connector.connect( 
  35.     host=mysql_host, 
  36.     database=mysql_database, 
  37.     user=mysql_user, 
  38.     password=mysql_password 
  39. ) 
  40. print(f"{bcolors.HEADER}Connection to MySQL Server succeeded.{bcolors.ENDC}") 
  41.  
  42. #MongoDB connection 
  43. print(f"{bcolors.HEADER}Connecting to MongoDB server...{bcolors.ENDC}") 
  44. myclient = pymongo.MongoClient(mongodb_host) 
  45. mydb = myclient[mongodb_dbname] 
  46. print(f"{bcolors.HEADER}Connection to MongoDB Server succeeded.{bcolors.ENDC}") 
  47.  
  48. print(f"{bcolors.HEADER}Database connections initialized successfully.{bcolors.ENDC}") 
  49.  
  50. #Start migration 
  51. print(f"{bcolors.HEADER}Migration started...{bcolors.ENDC}") 
  52. dblist = myclient.list_database_names() 
  53. if mongodb_dbname in dblist: 
  54.     print(f"{bcolors.OKBLUE}The database exists.{bcolors.ENDC}") 
  55. else: 
  56.     print(f"{bcolors.WARNING}The database does not exist, it is being created.{bcolors.ENDC}") 
  57.  
  58. #Function migrate_table  
  59. def migrate_table(db, col_name): 
  60.     mycursor = db.cursor(dictionary=True) 
  61.     mycursor.execute("SELECT * FROM " + col_name + ";") 
  62.     myresult = mycursor.fetchall() 
  63.  
  64.     mycol = mydb[col_name] 
  65.      
  66.     if delete_existing_documents: 
  67.         #delete all documents in the collection 
  68.         mycol.delete_many({}) 
  69.  
  70.     #insert the documents 
  71.     if len(myresult) > 0: 
  72.         x = mycol.insert_many(myresult) 
  73.         return len(x.inserted_ids) 
  74.     else: 
  75.         return 0 
  76.  
  77. #Iterate through the list of tables in the schema 
  78. table_list_cursor = mysqldb.cursor() 
  79. table_list_cursor.execute("SELECT table_name FROM information_schema.tables WHERE table_schema = %s ORDER BY table_name LIMIT 15;", (mysql_schema,)) 
  80. tables = table_list_cursor.fetchall() 
  81.  
  82. total_count = len(tables) 
  83. success_count = 0 
  84. fail_count = 0 
  85.  
  86. for table in tables: 
  87.     try: 
  88.         print(f"{bcolors.OKCYAN}Processing table: {table[0]}...{bcolors.ENDC}") 
  89.         inserted_count = migrate_table(mysqldb, table[0]) 
  90.         print(f"{bcolors.OKGREEN}Processing table: {table[0]} completed. {inserted_count} documents inserted.{bcolors.ENDC}") 
  91.         success_count += 1 
  92.     except Exception as e: 
  93.         print(f"{bcolors.FAIL} {e} {bcolors.ENDC}") 
  94.         fail_count += 1 
  95.          
  96. print("") 
  97. print("Migration completed.") 
  98. print(f"{bcolors.OKGREEN}{success_count} of {total_count} tables migrated successfully.{bcolors.ENDC}") 
  99. if fail_count > 0: 
  100.     print(f"{bcolors.FAIL}Migration of {fail_count} tables failed. See errors above.{bcolors.ENDC}") 
  101.  
  102. end_time = datetime.datetime.now() 
  103. print(f"{bcolors.HEADER}Script completed at: {end_time} {bcolors.ENDC}") 
  104. print(f"{bcolors.HEADER}Total execution time: {end_time-begin_time} {bcolors.ENDC}") 

警告

該腳本適用于中小型 MySQL 數(shù)據(jù)庫(kù)，它有幾百個(gè)表，每個(gè)表有幾千行。對(duì)于具有數(shù)百萬(wàn)行的大型數(shù)據(jù)庫(kù)，性能可能會(huì)受到影響。在開(kāi)始實(shí)際遷移之前，請(qǐng)?jiān)诒砹斜聿樵兒蛯?shí)際表select查詢上使用 LIMIT 關(guān)鍵字對(duì)有限行進(jìn)行檢測(cè)。

下載 

帶點(diǎn)擊此處從 GitHub 下載整個(gè)腳本：
https://github.com/zshameel/MySQL2MongoDB

【51CTO譯稿，合作站點(diǎn)轉(zhuǎn)載請(qǐng)注明原文譯者和出處為51CTO.com】

責(zé)任編輯：梁菲 來(lái)源： DZone
PythonMySQL MongoDB 數(shù)據(jù)遷移
相關(guān)推薦

2017-10-20 08:45:15

數(shù)據(jù)庫(kù)MongoDBMySQL

2016-12-12 19:16:43

數(shù)據(jù)云端

2022-08-29 14:14:22

云計(jì)算云遷移數(shù)據(jù)分析

2021-01-28 09:00:00

SQL數(shù)據(jù)庫(kù)NoSQL

2016-11-11 00:00:16

MySQLOracle數(shù)據(jù)

2013-10-09 09:35:49

數(shù)據(jù)中心遷移虛擬化

2013-10-09 09:54:46

數(shù)據(jù)中心云計(jì)算

2019-03-25 12:20:29

數(shù)據(jù)MySQL性能測(cè)試

2020-01-13 15:22:42

ERP云平臺(tái)遷移

2009-08-06 09:20:30

2010-08-12 09:43:31

CassandraMongoDB

2019-04-23 08:00:08

Azure微軟云遷移

2020-04-20 08:08:23

MongoDBElasticsear數(shù)據(jù)庫(kù)

2015-09-14 14:49:39

MySQLMariaDBLinux

2018-07-10 14:46:04

LinuxShellsudo

2020-12-02 14:38:21

SQL數(shù)據(jù)庫(kù)MySQL

2017-10-16 00:17:56

云計(jì)算信息管理遷移

2023-08-23 09:00:00

區(qū)塊鏈以太坊

2021-07-13 09:45:48

CentOSAlmaLinux命令

2018-08-15 15:15:00

數(shù)據(jù)基礎(chǔ)設(shè)施遷移
點(diǎn)贊
收藏

51CTO技術(shù)棧公眾號(hào)