小编Gio*_*ous的帖子

假设我有一个合奏的Zookeeper，它已启动并正在运行以促进和服务Apache Kafka（Confluent的发行版）。

3个实例（clientPorts： 2181，2182和2183）已经被配置并启动如下所示：

./bin/zookeeper-server-start etc/kafka/zookeeper.properties
./bin/zookeeper-server-start etc/kafka/zookeeper1.properties
./bin/zookeeper-server-start etc/kafka/zookeeper2.properties

在任何给定时间，如何检查哪个Zookeeper实例是集合的领导者？

假设我有一个多代理（在同一主机上运行）Kafka 设置，其中包含 3 个代理和 50 个主题，每个主题配置为具有 7 个分区和 3 的复制因子。

我有 50GB 的内存用于 kafka，并确保 Kafka 日志永远不会超过这个内存量，所以我想配置我的保留策略以防止这种情况。

我已经设置了删除清理策略：

log.cleaner.enable=true
log.cleanup.policy=delete

并且需要配置以下属性，以便每周删除数据并且永远不会耗尽内存：

log.retention.hours
log.retention.bytes
log.segment.bytes
log.retention.check.interval.ms
log.roll.hours

这些主题包含由总大小约为 10GB 的数据库表流式传输的数据（但插入、更新或删除在这些主题中不断流式传输）。

我应该如何配置上述参数，以便每 7 天删除一次数据，并确保在需要时可以在较短的窗口中删除数据，以免内存不足？

根据我的理解，分区和消费者被捆绑成 1:1 的关系，其中单个消费者处理分区。但是有没有这样一种在处理过程中重新分区的方法呢？我们目前正在尝试优化一个过程，在这个过程中，主题在一个组中被消费，但在某些情况下，某些消费者的数据处理需要更长的时间，而其他消费者已经空闲。它类似于数据清理，其中某个分区可能不再需要清理，而其他分区则需要模糊匹配，从而增加了消费者执行任务的复杂性。

我正在尝试为异常编写代码。如果输入与下面的模式不匹配（这只是示例），它将抛出异常消息。

8454T3477-90

这是我想出的代码。但是，我不确定这是否是正确的方法...

public void setLegalDescription(String legalDescription) throws MyInvalidLegalDescriptionException
    {
        String valid = ("[0-9999][A-Z][0-9999]-[0-99]");
        if (!legalDescription.matches(valid))
        {
            //throw new MyInvalidLegalDescriptionException("Invalid format! Should be " + "e.g 4050F8335-14");
        }
        this.legalDescription = legalDescription;
    }

在运行 kafka 消费者时，我收到以下错误：

Caused by: org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id 13
Caused by: org.apache.avro.AvroTypeException: Invalid default for field key_id: "null" not a ["null",{"type":"string","avro.java.string":"String"}]
    at org.apache.avro.Schema.validateDefault(Schema.java:1512)
    at org.apache.avro.Schema.access$300(Schema.java:86)
    at org.apache.avro.Schema$Field.<init>(Schema.java:493

我的 avro 架构是这样的：

{ "name": "key_id", "type": [ "null", "string" ], "default": "null"},

我尝试了多种选择来解决这个问题，但它仍然存在。

我需要更好地理解我的 LightGBM 模型，所以我使用 SHAP Tree 解释器。需要lightgbm对数据进行编码，我将相同的数据传递给树解释器。因此，我担心 SHAPTreeExplainer会将shap_values()我的数据视为数字数据。如何指定数据是分类的？这会改变 SHAP 值的计算吗？

我已经浏览过文档了。

像往常一样，看到拆分方法相对于其他方法的好处有点令人困惑。

1. 我看不到两者之间的区别/利弊
   • Topic1 -> P0 和 Topic 2 -> P0
   • 在Topic 1 -> P0, P1
     从2个主题或单一主题/ 2分区的消费拉动，而P0和P1将举行不同的事件类型或实体。

你唯一的好处是我可以看到另一个消费者是否需要主题 2 数据，那么它很容易消费

2. 关于主题自动生成，这种方式背后有什么好处还是一段时间后会失控？

谢谢

请帮助我启动并运行 Kafka 服务器，我可以很好地运行 Zookeeper，但是当我运行kafka-server-start.bat D:\Kafka\config\Server.properties命令控制台时没有返回任何输出。

注意：我已更改log.dirs为本地文件夹。

Kafka-命令输出

这是 server.properties 配置

broker.id=0
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=D:\kafka\kafkalogs
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=localhost:2181
zookeeper.connection.timeout.ms=6000
group.initial.rebalance.delay.ms=0

截至目前，我正在做类似这样的阅读 avsc 文件以获取架构

value_schema = avro.load('client.avsc')

我可以做些什么来使用主题名称从融合模式注册表中获取模式吗？

我找到了一种方法，但不知道如何使用它。

https://github.com/marcosschroh/python-schema-registry-client

来自 kafka 文档

当生产者将 acks 设置为“全部”（或“-1”）时，此 min.insync.replica 配置指定必须确认写入才能将写入视为成功的最小副本数。

它说当最小数量的同步副本确认时，写入成功但是当我使用min.insync.replica1 和 3运行性能测试时 （对于 5 个代理设置中的分区 = 1 和 RF = 5 的主题），kafka 的性能生产者，与acks='all'，相同。

那么，min.insync.replica每个主题的配置是否会影响 Kafka 生产者吞吐量（隔离运行）acks="all"？