apache-kafka 相关问题

Apache Kafka是一个分布式流媒体平台，用于存储和处理高吞吐量数据流。

如何使用 Benthos 读取和解码来自 Kafka 的 AVRO 消息及其关联的 kafka 密钥？

我正在使用 Benthos 从 Kafka 读取 AVRO 编码的消息，其中 kafka_key 元数据字段设置为还包含 AVRO 编码的有效负载。这些 AVRO 编码有效负载的模式存储...

go apache-kafka confluent-schema-registry benthos

回答 2 投票 0

考虑这个 KStream： KStream inputStream = StreamsBuilder.stream("kafka-topic", Consumed.with(Serdes.String(), Serdes.String())); 物化考虑这个 KStream： KStream<String, String> inputStream = streamsBuilder.stream("kafka-topic", Consumed.with(Serdes.String(), Serdes.String())); Materialized<String, List<String>, WindowStore<Bytes, byte[]>> with = Materialized.with(Serdes.String(), STRING_LIST_SERDE); KStream<Windowed<String>, List<String>> outputStream = inputStream .groupByKey() .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofSeconds(2))) .aggregate( ArrayList::new, (key, string, aggregate) -> { aggregate.add(string); return aggregate; }, with) .toStream(); outputStream 将聚合来自 inputStream info 的所有消息，定义时间范围内的消息数组。另外，现在我想将消息聚合到特定限制，例如直到列表大小不超过 50 条。如果列表在聚合过程中变得大于 50，我想以某种方式将其拆分为附加列表。基本上，我希望实现的输出是获取一组消息，其大小达到限制（例如 50 条），并且达到特定的时间范围，以先到者为准。为了实现这一目标，我在这里缺少什么？您可以尝试将 KTable 转换为 KStream 并执行 flatMapValues 来拆分列表，如下所示（Kotlin 中的代码）： val s = streamsBuilder.stream("kafka-topic", Consumed.with(Serdes.String(), Serdes.String())) val output = s .groupByKey() .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofSeconds(2))) .aggregate({ mutableListOf() }, { k: String, str: String, agg: List<String> -> agg.plus(str) }) .toStream() .flatMapValues { strList -> strList.chunked(50) } 但这意味着您将整个聚合列表加载到内存中 - 可能会也可能不会是一个问题，具体取决于列表大小和您的内存设置，但这绝对是需要记住的事情。

java spring spring-boot apache-kafka apache-kafka-streams

回答 1 投票 0

将 Kafka 扩展到 1 个数据中心内的安全区域

我的数据中心内有 2 个安全区域。我想部署kafka，这样如果Security zone-1中有Broker-1发布的主题，那么该主题也可以在Security zone-...

apache-kafka

回答 1 投票 0

Kafka Streams 用于聚合事件处理和对大量任务的可扩展性

我有以下任务：假设我正在开发一个由一百万用户使用的在线商店。用户在不同的时间进行购买。每个用户的每次购买都会变成一次交易事件。我...

apache-kafka apache-kafka-streams

回答 1 投票 0

Cypress 和 Kafka 自动化

我想知道是否有一些如何使用 cypress 和 apache kafka 进行自动化测试。为了能够在代理中“获取”特定主题，请发送主题（生产者）并接收主题（

javascript apache-kafka automation cypress

回答 1 投票 0

卡夫卡·卡夫|使用kafka-acls时没有配置Authorizer

使用 kafka-configs 创建一个 scram 用户： /bin/kafka-configs --bootstrap-server 代理：9092 --alter --add-config 'SCRAM-SHA-256=[password="password"]' --entity-type users --entity-name "

apache-kafka kraft

回答 1 投票 0

Kafka 连接 Amazon MSK

如何将 Kafka Connect 适配器与 Amazon MSK 结合使用？根据 AWS 文档，它支持 Kafka 连接，但没有记录如何设置适配器和使用它。

apache-kafka apache-kafka-connect amazon-msk

回答 3 投票 0

Kafka docker 撰写外部连接[重复]

我想将 9093 暴露到 docker 容器之外。当我将 kafka-0 端口设置为暴露给 9093 和 KAFKA_ADVERTISED_LISTENERS 时，我无法连接到 localhost:9093，如下所示...

docker apache-kafka docker-compose

回答 1 投票 0

Docker Kafka 与 Python 消费者

我正在使用dockerized Kafka并编写了一个Kafka消费者程序。当我在本地计算机上的 docker 和应用程序中运行 Kafka 时，它运行得很好。但是当我在