通过 Python 多处理队列执行 cassandra 插入查询

Posted 2023-03-22

【中文标题】通过 Python 多处理队列执行 cassandra 插入查询

【英文标题】：Executing a cassandra insert query through Python multiprocessing queue

【发布时间】：2015-08-14 11:07:32

【问题描述】：

我有一个 cassandra 键空间 sujata。我正在使用 python 驱动程序 cassandra.cluster 连接到 cassandra。sujata 的列族是 hello。 以下是我的代码：-

from multiprocessing import Process,Queue
from cassandra.cluster import Cluster
import os
queue=Queue()
cluster = Cluster(['127.0.0.1'])
metadata = cluster.metadata
session = cluster.connect("sujata")

def hi():
    global session
    global queue
    while True:
        y=queue.get()
        if y=="exit":
            os._exit(0)    
        else:
            print y
            session.execute(y)

if __name__=="__main__":
    x=Process(target=hi)
    x.start()
    for i in xrange(10):
        z="INSERT into hello(name) VALUES('" + str(i) + "');"
        queue.put(z)
        if i==9:
            queue.put("exit")
    session.cluster.shutdown()
    session.shutdown()

在表中，我有一个列name，我想在其中插入 i 的值。插入查询通过队列传递。我能够获取队列的内容。当我运行上面的代码时，输出为：-

INSERT into hello(name) VALUES('0');

session.execute() 不工作。 我无法理解为什么会这样。

【参考方案1】：

我没有 cassandra 机器，但我想，一旦你将连接部分移动到 prcoess-function hi()，它就会工作。喜欢：

def hi():
    cluster = Cluster(['127.0.0.1'])
    metadata = cluster.metadata
    session = cluster.connect("sujata")
    global queue
    while True:
        y=queue.get()
        if y=="exit":
            os._exit(0)    
        else:
            print y
            session.execute(y)

不知道为什么会这样，但我看到 global 变量在新进程中表现得很奇怪。

我猜这不是最好的方法。因为每次它都会连接到同一个数据库，这是多余的，需要关闭。希望有更好的答案。

编辑 1：

我没有正确阅读代码。您正在使用queue。所以进程hi 将只启动一次，queue 用于在进程之间进行通信。所以与数据库的连接也将只有一次。您不需要主进程中的数据库连接。所以将这部分转移到多进程函数是最好的方法。

【讨论】：

这工作正常。但正如你所说，我会等待更好的答案。

@sujata 从进程函数创建连接是一种可怕的反模式

@zehelvion 你能建议另一种方法吗？？

@RatDon 当然，没有“好”的方法可以做到这一点，但是如果您需要多个进程并希望所有进程都能够写入 Cassandra，您需要在有限数量的具有少量连接的进程，或者让进程通过管道连接到单个进程，该进程负责查询通过选择所有这些管道接受请求的数据库。

@zehelvion 这就是她做的对吗？从用于写入 cassandra 的单个进程创建连接，并且所有其他进程将通过管道（在本例中为 Queue）向该进程提供数据。