2

所以，我正在用 python 构建一个应用程序，它从 twitter 获取数据，然后将其保存到 cassandra。我目前的问题在于一个脚本，它从 kafka 读取数据并尝试将其写入 cassandra，如下所示：

import threading, logging, time
import multiprocessing
from cassandra.cluster import Cluster

from kafka import KafkaConsumer, KafkaProducer




class Consumer(multiprocessing.Process):
   def __init__(self):
        multiprocessing.Process.__init__(self)
        self.stop_event = multiprocessing.Event()

    def stop(self):
         self.stop_event.set()

    def run(self):
       consumer = KafkaConsumer(bootstrap_servers='localhost:9092',
                                 auto_offset_reset='earliest',
                                 consumer_timeout_ms=1000)
        consumer.subscribe(['twitter'])



    while not self.stop_event.is_set():
        for message in consumer:
            # session.execute(
            #     """
            #     INSERT INTO mensaje_73 (tweet)
            #     VALUES (message)
            #     """
            # )
            print(message)
            cluster = Cluster()
            session = cluster.connect('twitter')
            session.execute(
                    """
                    INSERT INTO mensaje_73 (tweet)
                    VALUES (message)
                    """
                )

            # if self.stop_event.is_set():
            #     break

    consumer.close()


   def main():

    tasks = [
        Consumer()
    ]

    for t in tasks:
        t.start()

    time.sleep(10)

    for task in tasks:
        task.stop()



if __name__ == "__main__":
     logging.basicConfig(
        format='%(asctime)s.%(msecs)s:%(name)s:%(thread)d:% 
   (levelname)s:%(process)d:%(message)s',
        level=logging.INFO
    )
    main()

我试图将测试消息插入到表 twitter.mensaje_73 中，它运行良好，如下所示：

import threading, logging, time
import multiprocessing
from cassandra.cluster import Cluster

from kafka import KafkaConsumer, KafkaProducer


cluster = Cluster()
session = cluster.connect('twitter')
session.execute(
    """
    INSERT INTO mensaje_73 (tweet)
    VALUES ('helooo')
    """
)

任何帮助将不胜感激:)

4

1 回答 1

2

所以这里的问题是，您的message变量在 CQL 中被视为文字，如果没有单引号，它将无法工作。因此，错误。

为了解决这个问题，我会使用准备好的语句，然后绑定message到它：

session = cluster.connect('twitter')
preparedTweetInsert = session.prepare(
        """
        INSERT INTO mensaje_73 (tweet)
        VALUES (?)
        """
    )
session.execute(preparedTweetInsert,[message])

试一试，看看它是否有帮助。

此外，这似乎是一个简单的数据模型。但是要问自己一件事，您将如何查询这些数据？tweet除非是您唯一的 PRIMARY KEY ，否则这是行不通的。这也意味着您可以查询单个推文的唯一方法是通过消息的确切文本。需要考虑一些事情，但按天分区可能是更好的选择，因为它会很好地分布并提供更好的查询模型。

于 2018-10-15T13:57:30.590 回答