卡夫卡消费者滞后

Question

我们一直在尝试创建一个 kafka 消费者，它试图在来自其他 kafka 集群的 60 个分区中以大约 2.7tb/小时的速度使用数据。

到目前为止，我们已经设法消耗了大约 2tb 的数据/小时并且无法赶上目标 (2.7)。

我们正在使用的集群具有存储问题的数据保留/删除率，因此我们需要在 3 分钟内使用该数据量。

详细信息， 我们在 6 台机器上使用数据，这些机器有 60 个分区。

import java.io.*;
import java.net.InetSocketAddress;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.time.Instant;
import java.util.*;
import javax.json.*;
import java.sql.Timestamp;

import org.apache.hadoop.HadoopIllegalArgumentException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import com.google.gson.Gson;
import com.google.gson.JsonElement;
import com.google.protobuf.util.JsonFormat;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.HAUtil;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.hadoop.security.UserGroupInformation;

public class NotificationConsumerThread implements Runnable {

    private final KafkaConsumer<byte[], byte[]> consumer;
    private final String topic;

    public NotificationConsumerThread(String brokers, String groupId, String topic) {
        Properties prop = createConsumerConfig(brokers, groupId);
        this.consumer = new KafkaConsumer<>(prop);
        this.topic = topic;
        this.consumer.subscribe(Arrays.asList(this.topic));
    }

    private static Properties createConsumerConfig(String brokers, String groupId) {
        Properties props = new Properties();
        props.put("bootstrap.servers", brokers);
        props.put("group.id", groupId);
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("session.timeout.ms", "120000");
        props.put("request.timeout.ms", "120001");
        props.put("max.poll.records", "280000");
        props.put("fetch.min.bytes", "1");
        props.put("max.partition.fetch.bytes", "10000000");
        props.put("auto.offset.reset", "latest");
        props.put("receive.buffer.bytes", "15000000");
        props.put("send.buffer.bytes", "1500000");
        props.put("heartbeat.interval.ms", "40000");
      //  props.put("max.poll.interval.ms", "420000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");

        return props;
    }



    @Override
    public void run() {
        try {
            Configuration confHadoop = new Configuration();
            
            confHadoop.addResource(new Path("redacted"));
            confHadoop.addResource(new Path("redacted"));
            confHadoop.setBoolean("dfs.support.append" ,true);


           
            confHadoop.set("fs.hdfs.impl", org.apache.hadoop.hdfs.DistributedFileSystem.class.getName());
            confHadoop.set("fs.file.impl", org.apache.hadoop.fs.LocalFileSystem.class.getName());
            confHadoop.set("hadoop.security.authentication","kerberos");
            confHadoop.set("dfs.namenode.kerberos.principal.pattern", "redacted");
            UserGroupInformation.setConfiguration(confHadoop); UserGroupInformation.loginUserFromKeytab("redacted", "redacted");

            FileSystem fileHadoop1 = FileSystem.get(confHadoop);
            StringBuffer jsonFormat3 = new StringBuffer();

            while (true) {
                String jsonFormat;
                String jsonFormat1;
                String jsonFormat2;


                DateFormat dateFormat = new SimpleDateFormat("yyyyMMddHH");
                dateFormat.toString();

                Date date = new Date();


                ConsumerRecords<byte[], byte[]> records = consumer.poll(3000);


                for (ConsumerRecord<byte[], byte[]> record : records) {


                    FlowOuterClass.Flow data = FlowOuterClass.Flow.parseFrom(record.value());
                    jsonFormat = JsonFormat.printer().print(data);
                    jsonFormat1 = jsonFormat.replaceAll("\\n", "");


                    JsonObject jsonObject1 = Json.createReader(new StringReader(jsonFormat1)).readObject();
                    Timestamp ts = new Timestamp(Long.parseLong(jsonObject1.getString("xxxx")));


                    date = new Date(ts.getTime());
                    jsonFormat2 = jsonFormat1.substring(0, jsonFormat1.length() - 1) + ", " + "\"xxxxx\"" + ": " + "\"" + dateFormat.format(date) + "\"" + "}\n";
                    jsonFormat3.append(jsonFormat2);

                }

                    String jsonFormat4 = jsonFormat3.toString();

                    if(jsonFormat4.length()>100000000) {
                        FSDataOutputStream stream = fileHadoop1.create(new Path("redacted-xxxxx" + dateFormat.format(date) + "/" + UUID.randomUUID().toString() + ".json"));
                        stream.write(jsonFormat4.getBytes());
                        stream.close();


                        
                        jsonFormat3.delete(0, jsonFormat3.length());
                    }
                }



        } catch (Exception e) {
            System.out.println(e);
        }
        consumer.close();
    }
}

这是滞后状态：

我们在互联网上找不到任何解决方案，因此我们很高兴了解如何使用 kafka 消费者使用这些大量数据的最佳实践。

谢谢！

Answer 1

您可以尝试做一些事情，看看您是否能够以最小的延迟赶上生成消息的原始速率。

1. 您应该增加消费者组中的消费者数量。根据您发布的图片，我可以看到有 10 个消费者在 6 台机器上运行。如果您的机器能够运行更多的消费者，那么您可能应该考虑增加消费者的数量。请注意，如果您可以将消费者数量增加到 12、15、20、30 中的任何一个，则效果会更好。这是因为我们希望所有消费者从主题中获得相同数量的分区。所以这个想法是消费者的数量应该是 60 倍。（您正在消费的主题中的分区数量）
2. 您试图通过将max.poll.records 更改为280000 来增加轮询时的记录数。请注意，当您以类似方式调整其他两个配置时，此配置只会起作用。您需要按比例更改max.partition.fetch.bytes 和fetch.max.bytes。我看到您已尝试将 max.partition.fetch.bytes 更改为 10000000(10MB) 大约。您还应该考虑调整此值fetch.max.bytes。因此，简而言之，您需要以正确的比例调整所有这些值。请仔细阅读，您可能会发现这很有用。 increase number of message in the poll
3. 如果上述两种方法都不起作用，这是您可以考虑的最后一步。因为我们知道 Kafka 中的分区决定了您可以实现的并行度。您可以考虑增加您正在消费的主题中的分区数量（将其更改为 120 或从当前 60 个分区中的更大数量）

我希望这会有所帮助。