使用Kafka事件的确切数量
Kafka是一个高性能、可扩展的分布式事件流平台，它用于实时处理、存储和传输大规模数据流。通过使用Kafka，可以轻松处理和管理事件流，并将其用于构建实时数据管道、流式处理应用程序和事件驱动的架构。
Kafka的核心概念包括Producer（生产者）、Consumer（消费者）、Topic（主题）、Partition（分区）和Offset（偏移量）。Producer负责将事件数据发布到Kafka集群中的特定Topic，而Consumer从特定的Topic中读取数据并进行处理。Topic将数据进行逻辑上的分组和分类，而Partition则将Topic分割成一系列有序的消息队列。每个消息在一个Partition中有唯一的Offset标识，Consumer通过指定Offset来消费数据。
使用Kafka事件可以提供以下优势：
- 可靠性：Kafka使用分布式提交日志（Distributed Commit Log）的方式来存储事件数据，确保数据的可靠性和持久性。即使在Producer或Consumer的故障情况下，数据仍然可以被安全地传输和处理。
- 可扩展性：Kafka的分布式架构和分区机制使得它可以处理大规模的事件数据流，并且可以根据需要进行水平扩展，以适应不断增长的数据量和吞吐量要求。
- 实时性：Kafka的设计目标之一是提供低延迟的事件处理能力。它支持实时数据传输和流式处理，使得用户能够几乎实时地获取和处理事件数据。
- 多样化的应用场景：Kafka在大数据领域有着广泛的应用，包括日志收集、用户行为追踪、指标监控、实时分析等。它还可以作为数据集成和数据传输的中间件，连接不同的应用系统和数据源。
腾讯云提供了Kafka的云托管服务，称为消息队列 CKafka。CKafka具有高可用性、高吞吐量、低延迟和可弹性扩展的特点。您可以通过腾讯云CKafka来搭建和管理Kafka集群，具体产品介绍和文档可以参考腾讯云官方网站：CKafka产品介绍。
相关·内容
1回答
致力于创建新的Kafka主题以分享一些信息，您能否帮助行业明智的最佳实践：
字段的数量？Kafka主题能有200个字段那么大吗？
要使用的消息格式( Avro/ JSON？)
保留天数
浏览 2提问于2019-01-08得票数 0
我们有一群生产者向一群消费者发送消息/事件。每条消息只能由一个消费者使用。我们知道，使用Kafka中的消费者组可以很容易地实现这种常见的场景。然而，我们也有一些额外的约束:不是每个消费者都可以使用每条消息。消息有附加的(任意)要求，只有满足这些要求的消费者才必须处理它们。对于消费者组来说，这仍然是可能的，其中消费者首先查看消息，如果消息不满足要求，则最终重新提交消息。然而，不能保证消息至少会被每个消费者看到一次，因此它们可能会无限期地反弹，尽管可能存在匹配的消费者。我们也不能设置多个主题，因为对使用者的要求是由用户而不是应用程序定义的任意复杂的布尔公式。这可能导致主题的组合爆炸。 此外，我们
浏览 26提问于2020-12-17得票数 0
使用Spring batch实现Kafka。开发了Spring boot应用程序后，我的Kafka生成器不断地产生消息。我想分批处理这些消息。但当我触发作业时，作业仍在持续运行。所以我决定在KafkaItemReader中添加pollTimeout。这样我就可以停止我的工作。但是在触发Job时，Kafka中会有多少条消息。我无法在谷歌中找到，如果我将pollTimeout设置为1000ms，那么KafkaItemReader中会有多少条消息。 提示会很有帮助 @ String>().partitions(0).consumerproperties(prop).name(“reader”
浏览 226提问于2021-03-16得票数 0
回答已采纳
1回答
我有以下场景。 我们在K8s的云中对我们的应用程序进行了负载测试。我们的入站消息是通过Kafka发送的，我们正在给Kafka回信。我们的架构是这样的: Kafka线程将消息推送到disruptor (阻塞等待策略，512大小)，业务线程将消息从disruptor中取出进行处理。为了模拟负载，我们在Kafka主题(4个分区)中设置了应用程序未运行时的关闭500K消息。然后，我们启动应用程序来测量负载。 我们看到的是，disruptor完全填满了剩余的0容量，然后开始耗尽，这种情况一次又一次地继续下去。 这是正确的行为，还是我们以错误的方式使用颠覆者？请分享你的想法？
浏览 35提问于2020-07-21得票数 0
我已经使用spring-kafka在Spring boot应用程序中使用SeekToCurrentErrorHandler配置了一个kafka消费者。我的消费者配置是：
@Bean
public ConsumerFactory<String, String> consumerFactory() {
Map<String, Object> props = new HashMap<>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafkaserver"
浏览 0提问于2019-09-10得票数 1
我正在评估vert.x框架，看看是否可以减少使用spring引导开发的微服务之间基于Kafka的通信。
问题是:我能代替
使用vert.x事件总线的Kafka和
基于vert.x的spring引导微服务
浏览 1提问于2018-12-19得票数 13
回答已采纳
1回答
我想问问消费者群体，但是在云之间..。
这是我的例子。我有5个云服务器，让我们说A，B，C，D，E，F
生产者:经纪人集群:B、C、消费者:d、E、F
同一群体的消费者:d、E
我想问，我们能用同一个组配置云D和云E吗？卡夫卡经纪人是如何从不同的云端注意到这个群体的？
因为当我开始我的消费脚本(我使用kafka/laravel)时，我只运行了kafka : my，然后是kafka消费者运行。
当我使用多个云时，我需要在每个云中运行每个脚本。我不知道我的代理发生了什么，代理集群能够了解消费者组吗？或者我需要对经纪人中的每一个消费群体进行分类？？
Broker集群是代理之间的复制，对吗？我可以为D
浏览 3提问于2022-01-10得票数 0
我正在尝试通过flume将数据从kafka放入hdfs中。kafka_producer每10秒发送一条消息。我想在hdfs上将所有消息收集到一个文件中。这是我使用的flume的配置，但它在hdfs上存储了许多文件(一个用于消息)：
agent1.sources.kafka-source.type = org.apache.flume.source.kafka.KafkaSource
agent1.sources.kafka-source.zookeeperConnect = localhost:2181
agent1.sources.kafka-source.topic = prova
age
浏览 40提问于2017-07-07得票数 1
我计划利用Azure事件中心中的所有32个分区。要求:每个分区的“有序”处理至关重要。问题:如果我将所有32个分区的TU(吞吐量单位)增加到最多20个，我将得到40 MB的出入口。假设我计算出需要500个并行客户端线程并行处理(EventProcessorClient)才能满足吞吐量需求。如何使用EventProcessorClient实现这种级别的并行性，同时满足我的“订购”要求？顺便说一下，在Kafka中，我可以在一个主题中创建500个分区，而Kafka允许每个分区只允许一个线程，以保证事件顺序。
浏览 2提问于2020-12-07得票数 0
2回答
卡夫卡流如何用于事件源？
、、、、
我读到了如何通过使用Apache作为事件代理来实现事件源。()
如果我们看一下这张图片，它会显示事件是如何被写入Kafka的，然后Kafka流被用来在数据库中创建视图。
我在这里的问题是，我们如何才能使用卡夫卡流实现这一目标？如果我正确的话，它是一个客户端库，所以我们需要使用它的东西，比如一个名为“聚合服务”的微服务。这是实现这种设计的正确方法吗？它的规模会很好吗？
浏览 8提问于2022-08-01得票数 0
有没有人能帮我理解一下这个kafka broker的监控指标？我试着阅读了文档和一些文章，但我不清楚如何监控这些指标，以避免对prod环境中的kafka代理造成任何伤害。在我的prod环境中，对于此指标，我看到峰值达到每秒5k，但随后又回到每秒2k甚至更低。
每条消息的字节数也是如此。
任何详细的解释都会很有帮助。
浏览 1提问于2018-02-25得票数 0
有谁可以在日志记录方面比较kafka、编年史队列和disruptor之间的底层设计和性能？似乎kafka拥有最多的用户，但不要回避GC。
浏览 0提问于2018-01-23得票数 6
我目前正在使用Kafka流式DSL抑制功能的日期窗口。我们可能会遇到这样一种情况，一些事件可能会很晚才发生，超过了宽限期。
根据kafka流媒体文档，这样的事件将被丢弃，这不适合window。
请帮帮我。
1)是否有可能在同一流程中获得此类丢弃的事件？
Apache flink确实提供了对这种非常晚的事件的控制，并想知道这种功能是否在流中可用。
2)考虑到数百万事件流经系统，如何使用DSL- suppress将间歇性聚合数据保存在内存中？
任何timeline kafka流媒体社区都将很快提供rockDB支持，以避免应用程序因内存不足而崩溃。
浏览 2提问于2019-05-15得票数 0
我正在使用Kafka0.8.2.1，运行一个带有200个分区和RF=3的主题，日志保留设置为大约1GB。
未知事件导致集群进入“协调器加载”或“组加载”状态。有几个信号表明了这一点:在FetchOffsetRequests期间，基于pykafka的消费者开始失败，错误代码为14 COORDINATOR_LOAD_IN_PROGRESS，用于某些分区子集。这些错误是在使用自协调器加载之前就存在的使用者组时触发的。在代理日志中，出现了这样的消息：
[2018-05...] ERROR Controller 17 epoch 20 initiated state change for partiti
浏览 0提问于2018-08-22得票数 1
1回答
我正在使用AWS MSK，我希望启用ACL，但是当ACL打开时，我无法创建一个主题。我正在使用命令行工具来执行所有操作。以下是我所做工作的总结：
operation=ALLCreate 创建一个新的集群创建一个主题--使用AdminClient为client1打开ACL和使用AdminClient (通过提供--引导-服务器选项)--这会给创建相同主题的超时exceptionRe-try --这会产生一个错误，说明使用AdminClient已经存在主题的列表主题--这不返回使用Zookeeper连接的topicsCreate主题--这可以使用Zookeeper返回<code>H 1
浏览 4提问于2019-09-30得票数 8
回答已采纳
2回答
我正在使用星火结构化流来读取卡夫卡主题的记录；我打算计算在火花readstream中每个“微批”中收到的记录数量
这是一个片段：
val kafka_df = sparkSession
.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "host:port")
.option("subscribe", "test-count")
.load()
我从文档中了解到，当一个kafka_df启动时(接下来)，streamin
浏览 0提问于2018-08-07得票数 2
根据阅读kafka连接文档：
config.storage.topic=connect-configs
bin/kafka-主题-创建-动物园管理员本地主机:2181-主题连接-信任-复制-因子3-分区1 -config cleanup.policy=compact
offset.storage.topic=connect-offsets
bin/kafka-主题-创建--动物园管理员本地主机:2181-主题连接-偏移-复制-因子3-分区50 -config cleanup.policy=compact
status.storage.topic=connect-status
bin/kafk
浏览 3提问于2022-05-17得票数 0
回答已采纳
1回答
卡夫卡对库伯奈特连接问题的影响
、、、、
我在连接Fluentd到Kafka以获得我正在处理的集中式日志记录PoC时遇到了问题。
我目前正在使用以下配置：
米尼库
弗伦特
fluent/fluentd-kubernetes-daemonset:v1.14.3-debian-kafka2-1.0 (码头)
配置:我在yaml中设置了FLUENT_KAFKA2_BROKERS=<INTERNAL KAFKA BOOTSTRAP IP>:9092和FLUENT_KAFKA2_DEFAULT_TOPIC=logs env，用于流畅的守护进程。
卡夫卡
我有点期待看到在Kafka消费者中出
浏览 3提问于2022-02-10得票数 1
1回答
我们可以为kafka主题指定自定义分割器。因此，kafka生产者可以根据特定的自定义算法将消息确定地发送到特定的分区。
现在的问题是，当我增加分区数量时，kafka将如何在新分区之间重新分配现有消息？或者Kafka不会将消息分发到新分区？有可能触发这种重新分配吗？如果是这样，那么kafka将如何了解自定义分区程序，因为这段代码驻留在生产者部分？
浏览 6提问于2020-09-15得票数 1
我提交了多个spark streaming作业，这些作业使用相同的group.id消耗同一个kafka topic，正如纯Kafka文档中所说的，拥有相同group.id的多个消费者将加入同一个消费组，并且kafka topic中分块的记录将被拆分成这些消费者。然而，正如我在我的工作中测试的那样，这两个spark streaming作业仍然使用相同的主题的所有分区(不是均匀分割的)，并且在who进程期间没有发生重新分区/异常。这里有人知道Spark是如何管理与纯kafka平台不同的kafka分区偏移的吗？也许这是因为zookeeper管理spark中的kafka偏移量，而不是kafka本身管
浏览 63提问于2021-07-30得票数 0
1回答
我们目前有一个用来与kafka交互的库。但是我们计划将这个库开发成一个单独的应用程序。其他应用程序将使用rest端点发送kafka消息。计划在此应用程序中使用vert.x，以使其非阻塞和快速。这是一个好的策略吗。我担心1) http将使它比kafka的TCP慢2)流可能不可能3)单点故障，但作为独立的应用程序发布管理，控制和支持将比目前容易得多。这是一个好的策略吗?以前有人做过这样的事情吗？有什么建议吗？
浏览 0提问于2018-11-07得票数 0
目前还不清楚你是否可以像在Flume中那样在Kafka中进行扇出(复制)。
我想让Kafka将数据保存到HDFS或S3，并将该数据的副本发送到Storm进行实时处理。Storm聚合/分析的输出将存储在Cassandra中。我看到一些实现将所有数据从Kafka流向Storm，然后从Storm输出两个数据。但是，我希望消除Storm对原始数据存储的依赖。
这个是可能的吗？您是否知道有类似这样的文档/示例/实现？
另外，Kafka对S3存储有很好的支持吗？
我看到了用于存储到HDFS的Camus --您只是通过cron运行此作业来不断地将数据从Kafka加载到HDFS吗？如果作业的第二个实例在前一个
浏览 2提问于2013-06-23得票数 3
回答已采纳
2回答
Flink流-延迟和吞吐量检测
、、、、
我正在尝试运行Flink流作业。我想确定流处理的吞吐量和延迟。我已经启动了Kafka代理服务器，并且有来自kafka.How的传入消息，我是否计算每秒的消息数(吞吐量)？(如rdd.count。有没有类似的方法来获取传入消息的数量？)
(完整的场景:我已经通过Producer将消息作为Json对象发送。我在Json对象中添加了一些信息，比如作为字符串的名称和System.currentTimeMills。流过程中，如何通过messageStream(DataStream)获取发送的json对象？)
提前谢谢。
代码：
/** *读取卡夫卡中的字符串，并将其打印到标准输出。*/
public s
浏览 1提问于2017-05-29得票数 0
像这样的ErrorMessage：
org.apache.kafka.streams.errors.StreamsException:现有的内部主题error-span-aggregate-stream-KTABLE-SUPPRESS-STATE-STORE-0000000004-changelog有无效的分区:预期: 16；实际: 20。在org.apache.kafka.streams.processor.internals.InternalTopicManager.validateTopics(InternalTopicManager.java:579) ~kafka-streams-3
浏览 14提问于2022-11-28得票数 0
1回答
我已经设置了一个麋鹿堆栈。对于logstash实例，它有两个输出，包括Kafka和elasticsearch。 对于elasticsearch的输出，我希望保留字段@timestamp。对于Kafka的输出，我想删除字段@timestamp。所以我不能删除过滤器中的field @timestamp。我只想在Kafka输出中去掉它。 我还没有找到这样的解决方案。 追加 尝试使用克隆插件： clone {
clones => ["kafka"]
id => ["kafka"]
remove_field => ["@timest
浏览 211提问于2020-03-18得票数 1
回答已采纳
1回答
我有一个卡夫卡流应用程序，阅读一个卡夫卡主题与5个分区。
然后对数据进行多次聚合/重新分区。
我试图在这个场景中找到关于线程数量的建议，但发现很难理解。编写如下：
您可以在输入Kafka主题分区时启动应用程序的多个线程。
这意味着在我的例子中，5个线程是有效线程的最大数量。
但是，声称重新分区会使最大有效线程数翻倍：
..。这个主题是用与源主题相同数目的分区自动创建的，这意味着我们的应用程序现在从16个分区读取8个线程，从而产生某种争用。
这对我来说也是合理的，因为卡夫卡流也必须阅读它所创造的内部主题。
那么，有效线程5分区的最大数目还是5*(重新分区)？
浏览 0提问于2019-03-25得票数 2
1回答
运行时
纱线簇式
应用程序
火花结构流
从Kafka主题中读取数据
卡夫卡主题
现在有4个分区的主题-for。(分区数目可以更改)
每1秒增加最多2000条主题记录。
我发现Kafka主题分区的数量与火花执行器的数量(1:1)相匹配。
所以，在我的例子中，到目前为止我所知道的四个火花执行者是我认为的解决方案。
但是我担心数据吞吐量-能保证2000 rec/秒吗？
是否有任何关于在火花结构化流中设置适当配置的指导或建议？
尤其是spark.executor.cores，spark.executor.instances或者关于遗嘱执行人的事情。
浏览 1提问于2019-04-02得票数 3
回答已采纳
3回答
本机配置如下：
来源: AWS源brokerTrigger：Broker: Kafka serviceKnative 服务: containerConcurrency =1(确保一次请求/pod)的服务
将1000条消息发送到SQS队列，本机服务只创建大约100到150个豆荚来处理所有请求。
我查了一下，发现Kafka broker一次只将大约100到180个事件发送到K本地服务。
因此，本机服务只创建大约1xx个荚来处理这些事件。
剩馀的事件将无法发送到本机服务，代理将重试发送。
卡夫卡代理中是否有任何配置来增加将同时发送到K本地服务的事件的数量？
FYI：
与MTchannelbased
浏览 19提问于2022-09-12得票数 1
回答已采纳
我在Kubernetes上运行Kafka Connect (8-16个节点，自动标度)。我总共定义了44个连接器，每个Kafka主题一个(每个主题一个分区)。这些主题是由Debezium / Postgresql编写的。有3个Kafka节点。每个连接器的tasks.max设置为4。大部分连接器(但不是每个连接器！)有一个(总是一个)失败的任务，由于java.lang.IllegalStateException:没有当前分配的分区-0。
这里不是卡夫卡专家，请注意;)我假设有3个Kafka节点，因此3个工作人员做得很好，而第四个任务没有任何连接，所以失败了。但是为什么有时有4个任务运行得很好呢？
浏览 3提问于2019-03-24得票数 0
回答已采纳
当源主题分区计数= 1时工作正常。如果我将分区增加到任何大于1的值，我会看到下面的错误。既适用于低级，也适用于DSL API。有什么建议吗？可能会遗漏什么？
org.apache.kafka.streams.errors.StreamsException: stream-thread [StreamThread-1] Failed to rebalance
at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:410)
at org.apach
浏览 3提问于2017-02-19得票数 6
回答已采纳
我正在使用kafka-python-2.0.1来消费来自kafka代理的数据。到目前为止，我只运行了一个消费者实例。我们每5分钟就会收到2M条记录。我注意到kafka-python不能以更快的速度读取数据以及时消费所有消息。我是kafka-python的新手，不确定如何实现才能更快地读取数据。我应该运行多个使用者吗？ consumer = KafkaConsumer(bootstrap_servers='<broker>',security_protocol='SASL_SSL', sasl_mechanism = 'GSSAPI'
浏览 31提问于2020-11-06得票数 0
1回答
我们一直在尝试修复不想使用数据的消费者。它每4秒轮询一次，通常只收到大约100条记录。最大拉力约为18000。然后，它将原始字节写入文件。都是快速操作。
但是消费者组呼叫显示如下
[xyz@AZ254PP01 kafka]$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --new-consumer --group postproc-red2
GROUP, TOPIC, PARTITION, CURRENT OFFSET, LOG END OFFSET, LAG, OWNER
postproc
浏览 0提问于2016-04-11得票数 0
我试图设计卡夫卡的消费者，我有一个障碍，如何设计的过程。我正在考虑两个选择：
1. Process records directly from Kafka.
2. Staging table write from Kafka and process records.
方法1:处理来自Kafka的关键消息：
• Read messages one at a time from Kafka & if no records to process break the loop (configurable messages to process)
• Execute business
浏览 7提问于2021-03-25得票数 2
回答已采纳
1回答
更新Kafka事件日志
、、、
我正在使用卡夫卡作为管道来存储分析数据，然后再将其发送到S3，并最终发送到Redshift。我正在考虑用Kafka存储数据的最佳架构，这样就可以轻松地将数据存储到数据仓库中。
问题是，我从三个单独的页面事件中获取数据：
当页面被请求时。
加载页面时
当页面卸载时
这些事件发生在不同的时间(通常在几秒钟之内，但距离对方最多几分钟/小时)。
我希望最终在我的数据仓库中存储一个关于网页视图的单一事件。例如，一个日志条目如下：
pageid=abcd-123456-abcde, site='yahoo.com' created='2015-03-09 15:
浏览 1提问于2015-03-09得票数 0
回答已采纳
卡夫卡中的demo.incoming话题堆积如山。
我正在使用。每秒大约有70个请求通过Scrapy Cluster REST api (Producer)提交给Kafka。蜘蛛可以非常快地完成爬行，因为redis中的队列保持在一个非常低的数字，大多数时间少于10个。但demo.incoming中的消息数量每秒都在堆积。这是我用来查看Kafka中demo.incoming主题消息数的命令
kafka-run-class.sh kafka.tools.GetOffsetShell \
--broker-list localhost:9092,kafka-statefulset-2:9092,ka
浏览 1提问于2019-07-07得票数 1
我有一个Spark集群，总共有17个executors。我已经将Spark 2.1与Kafka集成在一起，并从topic中读取数据，例如：
val df = spark
.readStream
.format("kafka")
.options("kafka.bootstrap.servers","localhost:9092")
.options("subscribe","test")
.load
现在我想知道，当我以集群模式提交我的spark应用程序时，将分配多少个执行者(总共17个)来监
浏览 0提问于2018-09-05得票数 2
1回答
我正在使用Kafka拓扑结构，有时在更改applicationId和/或clientId属性之后，我会收到一个特定的kafka流上的错误："Missing source topic stream.webshop.products.prices.5 durign assignment. Returning error INCOMPLETE_SOURCE_TOPIC_METADATA“。我已经在每个Kafka节点的create.topic=true中设置了server.properties属性，但是似乎没有创建这个流的主题。
以下是我的Kafka流拓扑：
package ro.or
浏览 0提问于2019-09-05得票数 5
回答已采纳
1回答
当我的服务的每个客户端都有自己的kafka使用者时，我会考虑架构模式。
例如，有一个带有一些复制因子和一些分区的events主题，我用它们来实现可伸缩性。给定客户端的所有事件都属于一个分区(我使用clientId作为分区键)。
每个客户端都有自己的offset。所以我的应用程序接口允许使用offset来获取客户端事件。
它的系统设计很好吗？或者，什么是正确的API设计来获取事件？
浏览 0提问于2019-03-01得票数 0
我正在尝试创建一个卡夫卡主题
$KAFKA_HOME/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic testing
但是，报告了以下错误：
Error while executing topic command : Not all brokers have rack information. Add --disable-rack-aware in command line to make replica assignment withou
浏览 1提问于2018-01-23得票数 1
回答已采纳
我正在做一个项目，在这个项目中，不同的生产者(每个生产者由另一个客户代表)可以向我的服务发送事件。
此服务负责接收这些事件并将其存储在中间Kafka主题中，稍后我们将获取和处理这些事件。
问题是，一个客户可能会淹没事件，并影响另一个客户的事件处理，我正在尝试找到一种最好的方法来在不同客户之间创建隔离级别！
到目前为止，我能够通过为每个客户创建不同的主题来解决这个问题。
尽管这个解决方案暂时解决了这个问题，但似乎Kafka并不能很好地处理大量的主题100k+随着我们的生产者(客户)数量的增长，我们开始体验到单个代理的受控重启需要几个小时。
有没有人能建议一种更好的方法来创建生产者之间的隔离级别？
浏览 0提问于2017-10-14得票数 3
在kafka中，我使用./kafka-topics.sh命令创建了主题。该命令如下所示
./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 8 --topic test
主题测试有8个分区，kafka broker中有server.properties配置，也有num.partitions参数，默认为1。
现在我的具体问题是，它不会在主题测试的分区中产生歧义。它将考虑主题创建时提到的分区或server.properties中的num.partition
浏览 3提问于2016-11-10得票数 3
几天来，我一直在努力制作一个非常简单的应用程序，其中包括一个kafka经纪人和一个在本地主机上运行的kafka制片人，我在google上阅读了所有类似问题的可能答案，但我仍然无法让它发挥作用。
这是我得到的错误：
INFO kafka.client.ClientUtils$ - Fetching metadata from broker id:0,host:localhost,port:9092 with correlation id 0 for 1 topic(s) Set(clicks)
[ProducerSendThread-] INFO kafka.producer.SyncPro
浏览 3提问于2016-03-24得票数 0
回答已采纳
我正在创建一个nodejs库，它读取kafka中的事件并将它们推入文件中。计划是将数据推送到文件到达，例如，100 or(或2000事件)。我的问题是：
已经有一个库可以这样做了吗？，如果没有，我正在考虑执行以下操作：
阅读卡夫卡的事件并对其进行改造。
2-将数据推入数组中。
3-让第二个线程从该数组读取并使用writeStream将其推送到文件中。
还有更好的选择吗？
谢谢,
吕斯
浏览 7提问于2022-07-03得票数 0
这是关于卡夫卡MirrorMaker工具。
我在两台机器上配置了卡夫卡。
来源：
目的地: vm仅位于源端
卡夫卡在来源和目的地都是同样版本的卡夫卡卡夫卡_2.11-0.9.0.0
在源和目的地，各自的动物园管理员和卡夫卡服务器正在运行。
使用MirrorMaker工具，我想从源到目的地复制/制作主题镜像。
下面是我使用的命令：
./bin/kafka-run-class.sh kafka.tools.MirrorMaker --consumer.config ./config/mirror_consumer.properties --producer.config .
浏览 7提问于2016-05-06得票数 2
1回答
在我的应用程序中，我获得使用kafka排队并保存在磁盘上的数据，以及从kafka获取这些数据并进行处理的使用者。但当我的消费者试图阅读卡夫卡的数据时，我得到的例外是：
2017-06-09 10:57:24,733 ERROR NetworkClient Uncaught error in request completion:
org.apache.kafka.common.KafkaException: Error deserializing key/value for partition TcpMessage-1 at offset 155884487
at org.apache.kaf
浏览 2提问于2017-06-09得票数 1
回答已采纳
1回答
研究表明，与redis相比，当Kafka和redis都充当消息代理时，Kafka在每秒的msgs数和每秒传输的mbs数方面都有更好的吞吐量。
最近一项名为“实时流处理管道中数据摄取平台的比较”的研究表明，对于1kB的有效负载，kafka可以处理的请求量是redis的3倍。
到底是什么让Kafka作为消息代理比redis更强大/更快呢？
浏览 8提问于2021-10-14得票数 0
我正在尝试对KafkaConsumer使用kerberos身份验证。 为此，我添加了以下属性。 props.put("security.protocol", "SASL_PLAINTEXT"); // Setting this means we try to look in jaas.conf
props.put("sasl.kerberos.service.name", "kafka"); 但是，我收到一个错误消息： java.lang.IllegalArgumentException: Could not f
浏览 166提问于2019-09-19得票数 0
回答已采纳
我正在构建一个UI，我想在其中显示kafka主题的日志大小以及每个消费者的阅读量。我可以从zookeeper读取消费者偏移量。获取主题日志大小的有效方法是什么？我可能需要更新详细信息，比如每10秒更新一次。我可以从jmx获取日志大小，也可以使用简单的消费者和最新偏移时间。这是最好的可用选项，还是有其他有效的选项？
浏览 5提问于2015-09-29得票数 3
在回答之后，我也尝试过这样做。问题是，似乎我的kafka (0.8.1.1版)包了不同的运行脚本。因此，我尝试适应并产生了以下的监督配置：
[program:kafka]
command=pidproxy /var/run/kafka.pid /opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/server.properties
user=kafka
autostart=true
autorestart=true
startsecs=10
startretries=999
log_stdout=true
log_stderr=true
浏览 3提问于2014-07-20得票数 3
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
云直播活动推荐
腾讯云开发者
