Trying to use Queue for inserting values in cassandra python

Question

I am using cassandra to store the data i get from client program. I have a server program where i run threads for inserting values but without a queue it doesn't work as expected but i tried using queue, not sure where i am going wrong. Can someone tell me how to access insert statement multiple times.

client program:

import socket
import threading
import client_to_orc
import json
import pyangbind.lib.pybindJSON as pybindJSON

def con_server(server_host, server_port, thread_num):
    server_address = (server_host, server_port)
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    #print(sock.getsockname())
    ---
    # sw is the packets code comes here

    ---
    packet=(pybindJSON.dumps(sw))     ----> Data in string format

    sock.sendto(packet.encode("Utf-8"), server_address)
    data = sock.recv(4096)    
    data = data.decode()
    print('Client:' + data)


if __name__ == "__main__":
    SERVER_ADDR = ("localhost", 4242)
    threads = []
    for thread_num in range(10):
        thread_args = (SERVER_ADDR[0], SERVER_ADDR[1], thread_num)
        t = threading.Thread(target=con_server, name='con_servr', args=thread_args)
        t.start()
        threads.append(t)

    for t in threads:
        t.join()

Edited server:

def insert_Vmac(msg):
    cluster = Cluster(contact_points=['172.17.0.2'])
    session = cluster.connect()
    id=1
    session.execute("""INSERT INTO mykeyspace.hello(PacketID, PacketValue)
    VALUES(%s,%s)""",
    (id,msg))

def talkToClient(ip):
    logging.info("sending 'clients we received your data' to %s",ip )

    sock=socket.socket(socket.AF_INET,socket.SOCK_DGRAM)
    sock.sendto("ok".encode('utf-8'), ip)

def listen_clients(Host,port):
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.bind((Host,port))
    client_list=[]
    while True:
        msgg, client = sock.recvfrom(1024)
        message = msgg.decode("utf-8")
        msg = json.dumps(message)

        print('connected with : ' + client[0]+ ':' + str(client[1]))
        for line in msg:
            cluster = Cluster(contact_points=['172.17.0.2'])
            session = cluster.connect()
            id = 0
            query = SimpleStatement("""INSERT INTO 
                  mykeyspace.hel(PacketID, PacketValue)
                  VALUES(%s,%s)""" %(id, msg))
             session.execute_async(query)
         print('hello inserted')
         t = threading.Thread(target=talkToClient, args=(client,))
         t.start()

if __name__=="__main__":
    t2 = ThreadWithReturnValue(target=listen_clients, args=("localhost", 4242,))
    t2.start()

The Edited code stores the data only once. But i am sending data for 10 times but it doesn't store all the data.

Simon Fontana Oscarsson · Accepted Answer · 2019-06-10 19:29:48Z

1

You're using the same id for all requests so each write will overwrite the previous entry. That's why you only retrieve one row when you read the data.

That's with the assumption that "id" in your case is the PRIMARY KEY in your table.

answered Jun 10, 2019 at 19:29

Simon Fontana Oscarsson

2,1441 gold badge17 silver badges21 bronze badges

Sign up to request clarification or add additional context in comments.

1 Comment

KSp Over a year ago

I also thought this is the problem, but I tried to use uuid.uuid(1) for generating unique IDs for each insert call but didn't work.

Collectives™ on Stack Overflow

Trying to use Queue for inserting values in cassandra python

1 Answer 1

1 Comment

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

1 Comment

Your Answer

Sign up or log in

Post as a guest

Related