CODEWITHSUNDEEP
pythoncassandraqueuemultiprocessing
wonder
wonder

Reputation: 903

Executing a cassandra insert query through Python multiprocessing queue

I have a cassandra keyspace sujata.I am connecting to cassandra using python driver cassandra.cluster.The column family of sujata is hello. Following is my code:-

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()

In the table, i have a column name to which i want to insert the value of i.The insert query is passed through a queue.I am able to get the contents of queue.when I run the above code,the output is:-

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

The session.execute() is not working. I am unable to understand why this is happening.

Upvotes: 1

Views: 611

Answers (1)

RatDon
RatDon

Reputation: 3543

I don't have a cassandra machine, but I guess, It'll work once you move the connection part to the prcoess-function hi(). Like:

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)

Don't know why it's behaving like this, but I've seen global variables behaving oddly with new processes.

This is not the best way to do I guess. Because each time it'll connect to the same database which is redundant and need to be closed. I hope there might be a better answer.

Edit 1:

I didn't read the code properly. You are using queue. So the process hi will start once only and queue is being used to communicate between the processes. So the connection to the database will also be once only. You don't need the database connection in main process. So shifting that part to the Multiprocess Function is the best way.

Upvotes: 0

Related Questions