Python产生不同的Kafka分区

如何解决Python产生不同的Kafka分区

我正在尝试通过经典的Twitter流示例学习Kafka。我正在尝试使用我的生产者将基于2个过滤器的Twitter数据流式传输到同一主题的不同分区。例如，Twitter数据将track ='Google'分配到一个分区，将track ='Apple'分配给另一个分区。

$ cat tst.awk
{
    rec = ""
    while ( match($0,/-?[0-9.]+@/) ) {
        tgt = substr($0,RSTART,RLENGTH-1)
        rec = rec substr($0,1,RSTART-1) sprintf("%.2f",tgt)
        $0  = substr($0,RSTART+RLENGTH-1)
    }
    print rec $0
}

$ awk -f tst.awk file
Name_1:0.12@123
Name_1:0.12@123Name_2:0.33@456
Name_1:0.12@123Name_2:0.33@456Name_3:0.25@895

如何添加另一个轨道并将数据流传输到另一个分区。

同样，我如何使消费者从特定分区消费。

void print_num(int i)
{
    std::cout << i << '\n';
}
 
int main()
{
    // store a free function
    std::function<void(int)> f_display = print_num;
    f_display(-9);

    // store a lambda
    std::function<void()> f_display_42 = []() { print_num(42); };
    f_display_42();
}

解决方法

Kafka在消息密钥上对数据进行分区。在给定的代码中，您仅将value传递给Producer消息，因此该键将为null，因此将在所有分区之间循环。

请参阅您的Kafka库的文档，以了解如何为每条消息提供密钥

经过研究，我能够解决此问题：

在生产者端，指定分区：

self.producer.send(topic_name,value=data,partition=0)

在消费者方面，

consumer = KafkaConsumer(
       bootstrap_servers=['localhost:9092'],auto_offset_reset='latest',enable_auto_commit=True,auto_commit_interval_ms =  5000,max_poll_records = 100,value_deserializer=lambda x: json.loads(x.decode('utf-8')))
consumer.assign([TopicPartition('trial',0)])