這段代碼使用了 Python 的 multiprocessing 模塊來實現真正的并行處理，繞過 Python 的全局解釋器鎖（GIL）限制，從而在多核 CPU 上并發(fā)執(zhí)行多個 SQL 語句。

from pyhive import hive
import multiprocessing
 
# 建立連接
conn = hive.Connection(host="localhost", port=10000, username="your_username", password="your_password")
 
# SQL 語句列表
sql_statements = [
    "INSERT INTO table1 VALUES (1, 'value1')",
    "INSERT INTO table1 VALUES (2, 'value2')",
    "INSERT INTO table1 VALUES (3, 'value3')"
]
 
# 定義執(zhí)行函數
def execute_sql(sql):
    with conn.cursor() as cursor:
        cursor.execute(sql)
 
# 確保多進程代碼只在主進程中執(zhí)行
if __name__ == '__main__':
 
    # 使用進程池并發(fā)執(zhí)行
    with multiprocessing.Pool() as pool:
        pool.map(execute_sql, sql_statements)
 
    # 關閉連接
    conn.close()

1. 導入模塊

from pyhive import hive
import multiprocessing

• pyhive: 這是用于連接和操作 Hive 數據庫的 Python 庫。hive.Connection 用于建立與 Hive 數據庫的連接。
• multiprocessing: 這是 Python 的標準庫，用于創(chuàng)建和管理進程。通過 multiprocessing，我們可以繞過 Python 的 GIL（全局解釋器鎖）限制，實現真正的并行處理。

2. 建立數據庫連接

conn = hive.Connection(host="localhost", port=10000, username="your_username", password="your_password")

• 這里我們使用 hive.Connection 建立一個到 Hive 數據庫的連接。
• 參數：
  • host: HiveServer2 的主機地址，通常是 localhost 或 HiveServer2 運行的服務器 IP。
  • port: HiveServer2 的端口號，默認是 10000。
  • username: 連接 Hive 使用的用戶名。
  • password: 連接 Hive 使用的密碼。

這個連接對象 conn 將在后續(xù)的代碼中用于創(chuàng)建游標（cursor），并通過游標執(zhí)行 SQL 語句。

3. 定義 SQL 語句列表

sql_statements = [
    "INSERT INTO table1 VALUES (1, 'value1')",
    "INSERT INTO table1 VALUES (2, 'value2')",
    "INSERT INTO table1 VALUES (3, 'value3')"
]

• 這里定義了一個包含多個 SQL 語句的列表 sql_statements。每個語句都是一個插入操作，將數據插入到 Hive 表 table1 中。
• 你可以根據實際需求修改這些 SQL 語句。

4. 定義執(zhí)行函數

def execute_sql(sql):
    with conn.cursor() as cursor:
        cursor.execute(sql)

• execute_sql 函數是用于執(zhí)行單個 SQL 語句的函數。
• with conn.cursor() as cursor：為當前數據庫連接創(chuàng)建一個游標對象 cursor，這個游標用于執(zhí)行 SQL 語句。
  • cursor.execute(sql)：執(zhí)行傳入的 SQL 語句。
• 這個函數會被進程池中的每個進程調用，每個進程都會獨立執(zhí)行一個 SQL 語句。

5. 使用進程池并發(fā)執(zhí)行

with multiprocessing.Pool() as pool:
    pool.map(execute_sql, sql_statements)

• multiprocessing.Pool()：創(chuàng)建一個進程池。進程池可以管理一組工作進程，并將任務分配給這些進程。
  • 默認情況下，Pool() 會根據系統的 CPU 核心數創(chuàng)建相應數量的工作進程。
  • 你可以通過參數指定池中的進程數量，例如 Pool(4) 表示創(chuàng)建 4 個工作進程。
• pool.map(execute_sql, sql_statements)：
  • pool.map 方法會將 execute_sql 函數應用到 sql_statements 列表中的每個元素上。
  • pool.map 方法會自動將 SQL 語句列表分配給進程池中的工作進程，每個進程獨立執(zhí)行一個 SQL 語句。
  • 這個過程是并行的，多個進程可以同時執(zhí)行不同的 SQL 語句，從而提高執(zhí)行效率。

6. 關閉數據庫連接

conn.close()

• 在所有 SQL 語句執(zhí)行完畢后，我們關閉數據庫連接，釋放資源。

進程池的工作原理

multiprocessing.Pool 提供了一種方便的方式來并行化執(zhí)行函數。其工作原理如下：

1. 創(chuàng)建進程池：當你創(chuàng)建一個 Pool 對象時，會啟動多個工作進程（數量可以指定，或默認根據 CPU 核心數決定）。
2. 任務分配：當你調用 pool.map 時，進程池會將任務（在這里是 execute_sql 函數）分配給空閑的工作進程。
3. 并行執(zhí)行：每個工作進程獨立執(zhí)行分配給它的任務，互不干擾。
4. 結果收集：pool.map 會收集所有工作進程的執(zhí)行結果，并按照原始任務列表的順序返回結果。

為什么使用進程池而不是線程池？

1. GIL 限制：Python 的全局解釋器鎖（GIL）限制了多線程的并行執(zhí)行能力，尤其是在 CPU 密集型任務中，多線程并不能充分利用多核 CPU。
2. 進程并行：multiprocessing 模塊通過創(chuàng)建多個進程來繞過 GIL 限制，每個進程都有自己的 Python 解釋器和內存空間，因此可以實現真正的并行執(zhí)行。
3. 適用場景：
   • 線程池：適合 I/O 密集型任務（例如，等待數據庫查詢結果）。
   • 進程池：適合 CPU 密集型任務（例如，并行計算、數據處理等），或者你需要繞過 GIL 限制時。

注意事項

1. 數據庫連接：在多進程環(huán)境中，每個進程都有自己的內存空間，因此每個進程需要獨立的數據庫連接。在上述代碼中，每個進程都通過 conn.cursor() 創(chuàng)建了自己的游標。
2. 進程開銷：創(chuàng)建和銷毀進程有一定的開銷，因此對于非常短小的任務，進程池可能不會顯著提高性能。在這種情況下，可以考慮調整進程池的大小或使用其他優(yōu)化手段。
3. 連接池：如果你的程序需要頻繁訪問數據庫，可以考慮使用數據庫連接池來復用數據庫連接，減少連接建立和關閉的開銷。

總結

• 進程池：通過 multiprocessing.Pool 實現，可以繞過 Python 的 GIL 限制，實現真正的并行處理。
• 適用場景：適合 CPU 密集型任務或需要并行執(zhí)行多個獨立任務的場景。
• 代碼結構：
  • 建立數據庫連接。
  • 定義 SQL 語句列表。
  • 定義執(zhí)行函數 execute_sql。
  • 使用進程池并發(fā)執(zhí)行 SQL 語句。
  • 關閉數據庫連接。

通過這種方式，你可以充分利用多核 CPU 的優(yōu)勢，并發(fā)執(zhí)行多個 SQL 語句，從而提高程序的執(zhí)行效率。

解決多進程報錯

你遇到的錯誤是 RuntimeError，這是因為你在使用 multiprocessing 時沒有正確地保護代碼的入口點。具體來說，在 Windows 系統上（以及其他非 fork 的啟動方式），你必須將多進程相關的代碼放在 if __name__ == '__main__': 語句塊中，以避免子進程在啟動時重新導入主模塊并執(zhí)行不必要的代碼。

錯誤原因：

在 Windows 系統中，Python 的 multiprocessing 模塊使用 spawn 啟動子進程，這意味著子進程會重新導入當前腳本。如果不加以保護，子進程會再次執(zhí)行主模塊中的代碼，導致遞歸創(chuàng)建進程并拋出錯誤。

解決方案：

你需要將多進程相關的代碼放在 if __name__ == '__main__': 語句塊中，確保只有主進程會執(zhí)行這些代碼，而子進程不會。

修改后的代碼：

import multiprocessing
 
data = [
    "1",
    "2",
    "3"
]
 
# 定義執(zhí)行函數
def print_str(data):
    print(data)
 
# 確保多進程代碼只在主進程中執(zhí)行
if __name__ == '__main__':
    # 使用進程池并發(fā)執(zhí)行
    with multiprocessing.Pool() as pool:
        pool.map(print_str, data)

解釋：

• if __name__ == '__main__': 確保了只有在直接運行當前腳本時，才會執(zhí)行其中的多進程代碼。子進程不會執(zhí)行這個代碼塊，從而避免了遞歸創(chuàng)建進程的問題。
• 在 Windows 系統上，這是使用 multiprocessing 時必須遵循的慣用寫法。

其他注意事項：

• 如果你打算將腳本打包成可執(zhí)行文件（例如使用 pyinstaller），你還需要調用 multiprocessing.freeze_support()，不過在大多數腳本運行的情況下，這個調用不是必須的。

例如：

if __name__ == '__main__':
    multiprocessing.freeze_support()  # 如果需要打包成可執(zhí)行文件，可以加上這行
    with multiprocessing.Pool() as pool:
        pool.map(print_str, data)

執(zhí)行sql 簡單示例

import multiprocessing
 
data = [  ]
 
# 定義執(zhí)行函數
def print_str(data):
    print(data)
 
# 確保多進程代碼只在主進程中執(zhí)行
if __name__ == '__main__':
 
    data2 = [
        "1",
        "2",
        "3"
    ]
 
    for i in data2:
        data_str = f"""
        inset into {i}
        """
        data.append(data_str)
 
 
    # 使用進程池并發(fā)執(zhí)行
    with multiprocessing.Pool() as pool:
        pool.map(print_str, data)

到此這篇關于Python使用進程池并發(fā)執(zhí)行SQL語句的操作代碼的文章就介紹到這了,更多相關Python進程池執(zhí)行SQL語句內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

最新評論