多个 Python 线程 – DuckDB 数据库

搜索快捷键 cmd + k | ctrl + k

文档 / 指南 / Python

多个 Python 线程

本页面演示了如何在多个 Python 线程中同时向 DuckDB 数据库插入数据和从其读取数据。这在有新数据不断流入且需要定期重新运行分析的场景中非常有用。请注意，这都发生在单个 Python 进程内（有关 DuckDB 并发性的详细信息，请参阅常见问题）。您可以随意按照此Google Colab notebook进行操作。

设置

首先，导入 DuckDB 和 Python 标准库中的几个模块。注意：如果使用 Pandas，请务必在脚本顶部添加import pandas（因为它必须在多线程之前导入）。然后连接到基于文件的 DuckDB 数据库，并创建一个示例表来存储插入的数据。此表将跟踪完成插入的线程名称，并使用DEFAULT 表达式自动插入发生插入时的时间戳。

import duckdb
from threading import Thread, current_thread
import random

duckdb_con = duckdb.connect('my_peristent_db.duckdb')
# Use connect without parameters for an in-memory database
# duckdb_con = duckdb.connect()
duckdb_con.execute("""
    CREATE OR REPLACE TABLE my_inserts (
        thread_name VARCHAR,
        insert_time TIMESTAMP DEFAULT current_timestamp
    )
""")

读写函数

接下来，定义将由写入线程和读取线程执行的函数。每个线程都必须使用.cursor()方法，基于原始连接创建到同一 DuckDB 文件的线程本地连接。此方法也适用于内存中的 DuckDB 数据库。

def write_from_thread(duckdb_con):
    # Create a DuckDB connection specifically for this thread
    local_con = duckdb_con.cursor()
    # Insert a row with the name of the thread. insert_time is auto-generated.
    thread_name = str(current_thread().name)
    result = local_con.execute("""
        INSERT INTO my_inserts (thread_name)
        VALUES (?)
    """, (thread_name,)).fetchall()

def read_from_thread(duckdb_con):
    # Create a DuckDB connection specifically for this thread
    local_con = duckdb_con.cursor()
    # Query the current row count
    thread_name = str(current_thread().name)
    results = local_con.execute("""
        SELECT
            ? AS thread_name,
            count(*) AS row_counter,
            current_timestamp
        FROM my_inserts
    """, (thread_name,)).fetchall()
    print(results)

创建线程

我们定义要使用的写入器和读取器数量，并定义一个列表来跟踪所有将要创建的线程。然后，首先创建写入线程，接着创建读取线程。接下来，打乱它们的顺序，以便它们以随机顺序启动，从而模拟同时进行的写入和读取操作。请注意，这些线程尚未执行，仅是定义了。

write_thread_count = 50
read_thread_count = 5
threads = []

# Create multiple writer and reader threads (in the same process)
# Pass in the same connection as an argument
for i in range(write_thread_count):
    threads.append(Thread(target = write_from_thread,
                            args = (duckdb_con,),
                            name = 'write_thread_' + str(i)))

for j in range(read_thread_count):
    threads.append(Thread(target = read_from_thread,
                            args = (duckdb_con,),
                            name = 'read_thread_' + str(j)))

# Shuffle the threads to simulate a mix of readers and writers
random.seed(6) # Set the seed to ensure consistent results when testing
random.shuffle(threads)

运行线程并显示结果

现在，启动所有线程并行运行，然后等待它们全部完成后再打印结果。请注意，由于随机化，读取器和写入器的时间戳如预期般交错显示。

# Kick off all threads in parallel
for thread in threads:
    thread.start()

# Ensure all threads complete before printing final results
for thread in threads:
    thread.join()

print(duckdb_con.execute("""
    SELECT *
    FROM my_inserts
    ORDER BY
        insert_time
""").df())

设置

读写函数

创建线程

运行线程并显示结果

关于此页面

本文中