这是通过kafka生产者读取消息并将其推送到主题的正确方法吗

Question

我已经编写了这个 Kafka 生产者并从桌面读取一个文件，然后将文件中的数据作为值推送，并通过每次读取每一行时添加一个来生成我自己的密钥。这是正确的方法还是我做了我不应该做的事情？请需要一些建议。 我可以在我的主题中看到消息，但每个消息都与一个键相关，所以如果我有一个用例，如果我从外部读取它，我可以推送任何这样的日志数据。我可以使用日志数据作为值，还是应该采用完全不同的逻辑。 请帮忙

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.util.Properties;

import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;


public class SyncProducer {

    public static void main(String[] args) throws IOException {

        File file = new File("/Users/adityaverma/Desktop/ParseData.txt");

        BufferedReader br = new BufferedReader(new FileReader(file));

        Properties properties  = new Properties();
        properties.setProperty("bootstrap.servers","127.0.0.1:9092");
        properties.setProperty("key.serializer",StringSerializer.class.getName()); // our key and values are String
        properties.setProperty("value.serializer",StringSerializer.class.getName());
        properties.setProperty("acks", "1"); 
        properties.setProperty("retries", "3"); 
        properties.setProperty("linger.ms", "1"); 



        Producer<String,String> producer = new org.apache.kafka.clients.producer.KafkaProducer<String,String>(properties);
        // these will go in random partition as we increment the key

        String line = " ";

        int key = 0;
        while((line = br.readLine()) != null){
        //  System.out.println(line);

        ProducerRecord<String,String> producerRecord = new ProducerRecord<String,String>("try_Buffered3Part",Integer.toString(key),line);
         key++;
         System.out.println(key);
        producer.send(producerRecord);

        }
        producer.close();
        System.out.println("exit");
    }

}

Answer 1

看起来不错。如果你对随机分区没问题，你可以选择使用空键。

您可能还想查看 logstash 和 kafka 的集成。

https://www.elastic.co/guide/en/logstash/current/plugins-outputs-kafka.html

Answer 2

是不是正确的方式

不清楚你的目标。你可以从终端消费数据吗？那你生产得很好。

您可以使用整数作为键。 Kafka 有一个 IntegerSerializer

使用 null 作为键或排除该参数是向随机分区发送数据的标准方式，并且不会遇到整数过载

我想通过Kafka从某个来源读取日志，然后将其写入HDFS。

如果您只想将数据记录到 Hadoop 中，Fluentd 或 Logstash 可以做到这一点。

在开始使用 Kafka 走这条路之前，您绝对应该选择一种数据格式。例如，Hadoop 和 Kafka 更喜欢 Avro 或 JSON 而不是 CSV。 Confluent 有大量关于将 Avro 生成到 Kafka 中的文档

您可以使用 Kafka Connect HDFS 连接器或 Apache Nifi 将 Kafka 数据导入 Hadoop。不要重新发明轮子来编写自己的消费者。