這段代碼使用了 Python 的 multiprocessing 模塊來實現(xiàn)真正的并行處理，繞過 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í)行函數(shù)
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)
 
    # 關(guān)閉連接
    conn.close()

1. 導(dǎo)入模塊

from pyhive import hive
import multiprocessing

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

2. 建立數(shù)據(jù)庫連接

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

• 這里我們使用 hive.Connection 建立一個到 Hive 數(shù)據(jù)庫的連接。
• 參數(shù)：
  • host: HiveServer2 的主機地址，通常是 localhost 或 HiveServer2 運行的服務(wù)器 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。每個語句都是一個插入操作，將數(shù)據(jù)插入到 Hive 表 table1 中。
• 你可以根據(jù)實際需求修改這些 SQL 語句。

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

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

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

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

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

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

6. 關(guān)閉數(shù)據(jù)庫連接

conn.close()

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

進程池的工作原理

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

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

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

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

注意事項

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

總結(jié)

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

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

解決多進程報錯

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

錯誤原因：

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

解決方案：

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

修改后的代碼：

import multiprocessing
 
data = [
    "1",
    "2",
    "3"
]
 
# 定義執(zhí)行函數(shù)
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 系統(tǒng)上，這是使用 multiprocessing 時必須遵循的慣用寫法。

其他注意事項：

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

例如：

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

執(zhí)行sql 簡單示例

import multiprocessing
 
data = [  ]
 
# 定義執(zhí)行函數(shù)
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)

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

最新評論