如何在基于kafka流的应用中降低延迟
在基于Kafka流的应用中降低延迟的方法有以下几个方面：
- 优化生产者端：
- 使用异步发送消息的方式，将消息发送到Kafka集群后立即返回，不等待服务器的响应，以减少网络延迟。
- 批量发送消息，将多个消息打包成一个批次发送，减少网络传输次数，提高吞吐量。
- 调整生产者的缓冲区大小，确保缓冲区足够大以容纳更多的消息，减少频繁的网络传输。
- 优化消费者端：
- 使用多线程消费消息，提高并发处理能力，减少消息处理的延迟。
- 提前预取消息，通过调整消费者的拉取大小和拉取间隔，提前获取更多的消息，减少等待时间。
- 使用消息过滤机制，只消费感兴趣的消息，减少不必要的消息处理。
- 调整Kafka集群配置：
- 增加分区数，将负载均匀分布在更多的分区上，提高并发处理能力。
- 调整副本数，增加副本数可以提高消息的可靠性，但也会增加网络传输的延迟，需要根据实际需求进行权衡。
- 调整Kafka的参数，如消息的最大大小、请求的最大大小等，根据实际情况进行调整，以提高性能。
- 使用合适的硬件设备：
- 使用高性能的网络设备，如千兆以太网、万兆以太网等，提高网络传输速度。
- 使用高性能的存储设备，如SSD硬盘，提高读写速度。
- 使用合适的压缩算法：
- Kafka支持多种压缩算法，如Gzip、Snappy、LZ4等，选择合适的压缩算法可以减少消息的传输大小，提高传输效率。
总结起来，降低基于Kafka流的应用延迟的方法包括优化生产者端和消费者端的处理逻辑，调整Kafka集群的配置，使用合适的硬件设备和压缩算法。通过综合考虑这些因素，可以有效地降低延迟，提高应用的性能和响应速度。
腾讯云相关产品推荐：
- 腾讯云消息队列 CKafka：https://cloud.tencent.com/product/ckafka
- 腾讯云云服务器 CVM：https://cloud.tencent.com/product/cvm
- 腾讯云云数据库 CDB：https://cloud.tencent.com/product/cdb
- 腾讯云云存储 COS：https://cloud.tencent.com/product/cos
相关·内容
当kafka-streams应用程序运行，Kafka突然关闭时，应用程序进入“等待”模式，用户和生产者线程发送警告日志给他们无法连接，当Kafka回来时，一切都应该(理论上)恢复正常。我正试图对这种情况发出警告，但我无法找到捕捉它并发送日志/度量的位置。我尝试了以下几点：
streams.setUncaughtExceptionHandler，但是这只发生在异常(这里不是这样)，，，扩展ProductionExceptionHandler，并将default.production.exception.handler属性更改为我的类，该类扩展了这个接口。再一次，和setUncaughtExcep
浏览 2提问于2020-08-23得票数 5
回答已采纳
1回答
如何控制Flink检查点的大小
、、、、
我正在运行一个简单的Flink聚合作业，它从Kafka消耗，并应用多个窗口(1小时、2小时hr...upto )和特定的滑动间隔，并在窗口上进行聚合。有时工作会重新启动，当它从最新的Kafka数据中启动窗口时，我们就会丢失数据。为了克服这一问题，我们启用了检查点，并且我看到检查点的大小正在增加( HashMapStateBackend具有Hdfs存储)。对于一个永久运行的Flink作业来说，什么是最好的检查方法?我们能控制检查点的大小吗?几天后它会很大吗？
尝试启用带有Hdfs存储的HashMapStateBackend检查指向。
浏览 10提问于2022-11-03得票数 0
1回答
我刚刚从Kafka和Kafka流媒体应用程序开始。我编写了一个，它从一个主题中消耗，处理这些消息，然后将它们发送到另一个主题。
据我所知，我发现运行这个Kafka Stream应用程序编码的唯一方法是：
从IDE运行Java类。
生成*.jar文件并从提示符中运行它。
我想知道是否有任何方式自动运行卡夫卡流媒体应用程序的卡夫卡服务器启动。例如:将*.jar文件复制到卡夫卡安装的某个文件夹，并在启动卡夫卡服务器时自动运行此流应用程序。
浏览 1提问于2017-12-19得票数 4
回答已采纳
我正在将Spring应用程序升级到Boot2.6.1、Cloud2021.0.0和CloudStream3.2.1。
这个应用程序有一堆KStreams，如：
@Bean
fun processEvent() =
Function<KStream<ID, EventReceived>, KStream<Id, UpdatedInfo?>> { events ->
...
}
现在，应用程序在启动时崩溃：
Caused by: org.springframework.kafka.Ka
浏览 25提问于2021-12-16得票数 2
回答已采纳
我的应用程序的日志文件不断累积在服务器上。我想通过Kafka将它们转储到HDFS中。我希望Kafka生产者读取日志文件，将它们发送到Kafka broker，然后将这些文件移动到另一个folder.Can。另外，有没有可能在Kafka producer中有复制逻辑？
浏览 1提问于2015-04-21得票数 2
我有一个Kafka流应用程序，它从几个主题中获取数据，并将数据加入到另一个主题中。
Kafka配置：
5 kafka brokers
Kafka Topics - 15 partitions and 3 replication factor.
注:我在卡夫卡经纪人运行的同一台机器上运行卡夫卡流应用程序。
每小时消耗/产生数以百万计的记录。每当我把任何卡夫卡经纪人打倒，它就进入再平衡，它需要大约。30分钟，有时甚至更多的再平衡，并多次它杀死了许多卡夫卡流的进程。
浏览 0提问于2017-09-12得票数 2
偶尔，我的kafka streams应用程序会因以下错误而死亡：
[-StreamThread-4] o.a.k.s.p.i.AssignedStreamsTasks : Failed to commit stream task 0_9 due to the
following error:
org.apache.kafka.common.errors.TimeoutException: Timeout of 60000ms expired before successfully
committing offsets {my-topic-9=OffsetAndMetadata{offset=5
浏览 2提问于2020-04-03得票数 1
关于源连接器是如何在KSQLDb和Kafka中工作的，我没有足够的信息。
数据填充到卡夫卡主题的速度有多快？
，如果KsqlDb流需要来自源的数据来连接数据，但数据仍然被加载了呢？
源连接器是否将更新/插入的数据发送到主题，它会立即发生？
你能帮我解决这些问题或提供一个好的教程，在那里我可以学到更多。
浏览 1提问于2020-07-12得票数 0
回答已采纳
我有这样的情景。
有4个主题：处理主题、异常主题、重试主题和拒绝主题。我有一个使用Kstream的处理器的spring云流应用程序。此处理器从异常主题读取消息，并基于每个消息中可用的标志创建两个kstream分支，用于重试主题和拒绝主题。现在需要做的是，在retry主题中出现的任何消息都必须等待一段特定的时间才能将其推回处理主题。谁能帮助我在春季云流中的kafka流应用程序中进行最好的设计或解决方案？能用磁通和Mono异步机构进行设计吗？任何资源或指导都将是很大的帮助。谢谢
浏览 2提问于2021-06-21得票数 0
2回答
我在我的Spring项目中使用Kafka-streams-binder。Kafka流应用程序使用6分钟的滑动窗口对结果进行汇总和分析。但问题是聚合操作正在产生重复的结果。
我想要抑制中间结果，并且只在应用程序的窗口结束后才发布。可以通过Kafka 2.1.1中的Kafka .supress()操作来实现。但是Spring版本没有最新的kafka使用这种功能。
项目依赖项
<spring-boot.version>2.1.9.RELEASE</spring-boot.version> <spring-cloud.version>Greenwich.SR3&l
浏览 2提问于2019-11-29得票数 2
1回答
我想使用Kafka流，但也与外部数据库和API交互。我读到过有人建议使用Kafka Connect这样做，但我想知道怎么做和为什么。即使没有Kafka流，也可以查询数据库(即使它没有Kafka连接连接器)和每个记录的API吗？或者我应该为这个用例使用另一个处理引擎？
浏览 3提问于2022-01-15得票数 0
我正在编写一个Flink应用程序，它使用kafka主题中的时间序列数据。时间序列数据包含指标名称、标记键值对、时间戳和值等组件。我已经创建了一个滚动窗口，用于根据指标键(指标名称、键值对和时间戳的组合)聚合数据。下面是主流看起来像 kafka source -> Flat Map which parses and emits Metric -> Key by metric
key -> Tumbling window of 60 seconds -> Aggregate the data -> write to the
sink. 我还想检查上面窗口之外是否
浏览 17提问于2021-08-09得票数 0
回答已采纳
2回答
火花流不从本地目录读取
、、、
我正在尝试使用spark编写一个星星之流应用程序。
应用程序应该从本地目录读取文本文件，并将其发送到Kafka集群。
当提交python脚本来激发引擎时，什么也没有发送给kafka。
我试图打印这些事件，而不是把它寄给卡夫卡，结果发现没有什么可读的。
这是脚本的代码。
#!/usr/lib/python
# -*- coding: utf-8 -*-
from pyspark import SparkContext, SparkConf
from pyspark.streaming import StreamingContext
from kafka import KafkaProducer
浏览 2提问于2016-04-17得票数 1
我正在使用Spring Cloud Stream (3.0.4.RELEASE)和Kafka-Streams绑定器(3.0.0.RELEASE)。我还使用了“函数式编程模型”(所以没有@StreamListener等)。多么可爱的一项技术！ 我需要能够在一天中的特定时间暂停新事件的流处理/消费。这会为事件创建一个“封锁期”。在‘封锁期’结束后，我将恢复流处理。因此，我希望能够使用代码暂停或打开/关闭KStream使用者。我似乎做不到！ 到目前为止，我已经尝试了什么？-使用/actuator/bindings端点启动/停止kafka-streams绑定。似乎这对于kafka-streams绑定
浏览 67提问于2020-06-16得票数 1
我有一个Spark结构的流媒体应用程序。应用程序从kafka接收数据，并且应该使用这些值作为参数来处理来自cassandra数据库的数据。我的问题是，我如何使用输入数据帧(kafka)中的数据，作为cassandra中的"where“参数"select”，而不会出现以下错误：
Exception in thread "main" org.apache.spark.sql.AnalysisException: Queries with streaming sources must be executed with writeStream.start();
这是
浏览 25提问于2021-05-24得票数 0
回答已采纳
我想用创建一个Kafka流应用程序，它集成了两个不同的Kafka集群/设置。我尝试使用多绑定器配置来实现它，如文档中提到的那样，并且类似于下面的示例：。
给定这样一个简单的函数：
@Bean
public Function<KStream<String, AnalyticsEvent>, KStream<String, UpdateEvent>> analyticsEventProcessor() {
return input -> input
.filter(new AnalyticsPredicate())
浏览 2提问于2022-08-10得票数 0
回答已采纳
在我的应用程序中，我将对我的Kafka集群执行某种健康检查。
目前，我创建了一个TopicMetadataRequest来检测死掉的代理：
Future {
// this will fail if Kafka is unavailable
consumer.send(new TopicMetadataRequest(Seq("health-check-topic"), 1))
}
不幸的是，由于集群拓扑/设置的原因，此调用会产生巨大的网络流量。
有没有更好的方法来检查kafka经纪人？我需要的是像true/false指示器这样简单的东西。
浏览 1提问于2015-07-17得票数 10
回答已采纳
1回答
流在Hadoop中的应用
、、、、
我正在尝试学习的例子。在这个例子中，我确实看到他们创建了一个单独的应用程序(类似于流应用程序)，该应用程序生成和消费数据，并在数据上应用模式匹配。他们现在还没有在.Till(如卡夫卡)之间放置一个流层，单应用程序就足以满足这一目的，这使得它非常优化。现在，我知道，如果我使用Kafka，那么我需要两个应用程序；一个用于将数据摄取到Kafka主题中，另一个用于消费Kafka主题的数据。我有几个问题我没有得到回答：
为什么在这个例子中他们没有使用任何流层(比如Kafka) ?？
什么时候?在哪里需要流？？
参考Flink CEP的例子，我想知道流层(如Kafka/Kinesis)将如何发
浏览 2提问于2016-06-13得票数 0
到目前为止，我所理解的是，Flink中有三种处理后期数据的方法：
删除延迟事件(这是事件时窗口操作符的默认行为)。因此，延迟到达的元素将不会创建新窗口。)。
重定向晚期事件(晚期事件也可以使用侧输出功能重定向到另一个DataStream )。
通过包含延迟事件(重新计算不完整的结果并发出更新)来更新结果
我不太明白对于非窗口操作符的延迟事件发生了什么，特别是当时间戳被分配到源时。这里我有一个FlinkKafkaConsumer：
new FlinkKafkaConsumer(
liveTopic,
deserializer,
conf
浏览 0提问于2021-05-28得票数 1
大多数文章都在读/写吞吐量方面比ActiveMQ等其他消息代理(MB)更好地描述了Kafka。在偏置的帮助下，理解读/写使它更快。但我不清楚抵消如何使它更快？
在阅读了Kafka架构之后，我对Kafka的可伸缩性和高吞吐量有了一些了解，但不清楚的原因如下：
使用偏移量，客户端可能知道需要读取的确切消息--这可能是使其性能更高的因素之一。
在其他MB的情况下，代理需要在使用者之间进行协调，以便消息只传递给使用者。但这种情况只适用于队列，而不适用于主题。那么是什么使卡夫卡的主题比其他MB的主题更快。
Kafka为可伸缩性提供分区，但其他消息代理(MB)，如ActiveMQ，也提供集群。那么
浏览 8提问于2017-06-19得票数 8
回答已采纳
Kafka是很常见的。所以很多公司都在使用它。我完全理解Kafka和Spark是如何工作的，我对他们都很有经验。我不理解的是用例。为什么你要把Kafka和Spark一起使用，而不仅仅是Spark呢？
在我看来，Kafka的主要用途是作为ETL管道中的中转区，用于实时(流)数据。
我假设有一个数据源集群，数据最初存储在其中。例如，它可以是Vertica、Cassandra、Hadoop等。
然后是一个处理集群，它从数据源集群读取数据，并将其写入分布式Kafka日志，这基本上是一个登台数据集群。
然后是另一个处理集群- Spark集群，它从Kafka读取数据，对数据进行一些转换和聚合，并将其写入最
浏览 3提问于2019-06-17得票数 3
2回答
我有一个将事件推送到kafka的设置，然后在同一个集群上运行应用程序。公平地说，扩展Kafka Streams应用程序的唯一方法是通过添加节点或增加Partitions来扩展kafka集群本身。
在这种情况下，我如何确保我的使用者不会关闭集群，并确保关键管道始终是"on"。是否有Topology Priority的概念可以避免可能的停机？我希望能够公开流，任何人都可以构建应用程序，而不损害核心管道。如果解决方案是设置另一个kafka集群，那么对所有特殊查询使用Apache storm更有意义吗？(据我所知，许多消费者仍然可能导致kafka集群的问题，但至少现在topology处
浏览 9提问于2017-01-25得票数 2
回答已采纳
1回答
我正在使用弹簧云流卡夫卡同步生产者在一个弹簧引导微服务。每次我们部署该服务时，对kafka的第一次调用就需要超过20秒才能将消息发布到主题。但随后的所有调用几乎不需要3到4毫秒。此问题也是随机发生的，是间歇性的，但大多数发生在我们重新启动服务时。我们使用kafka版本0.9.0.1和gradle依赖项，如下所示
compile('org.springframework.cloud:spring-cloud-starter-stream-kafka')
}
dependencyManagement {
imports {
mavenBom "org.springframe
浏览 2提问于2017-05-25得票数 1
1回答
我的Spring Cloud Stream应用程序中有一个断路器。当电路正在改变状态，并且我的流消费者是匿名的(不在组中)时，它可以很好地暂停/恢复流。
当消费者属于一个组时，暂停流可以很好地工作，但是恢复会被“忽略”，最终以超时结束并离开组。有没有解释为什么会出现这种不一致的行为？
Spring云流版本为3.0.8.RELEASE。
这是我的断路器状态转换处理程序：
@Component
public class CircuitBreakerKafkaStream {
private final Logger log = LoggerFactory.getLogger(CircuitB
浏览 0提问于2021-01-23得票数 0
2回答
我有两个流： topicA -> processingA -> topicB -> processingB -> topicC 登录我的应用程序时，我注意到在将processingA的输出发送到topicB和从topicB为processingB挑选消息之间，每次花费的时间超过100ms (相当于150ms)。它可能不是很多，但它会积累起来，最后相当简单的级联处理几乎只需要几秒钟。 我可以调优kafka使这些延迟尽可能接近于零吗？哪些配置参数对这些延迟有影响？ 我有大部分的默认配置。是不是commit.interval.ms导致了延迟？我已经将其从更高的默认值更改为.
浏览 14提问于2019-06-11得票数 5
3回答
我最近开始学习卡夫卡，最后得到了这些问题。
消费者和流的区别是什么？对我来说，如果有任何工具/应用程序使用来自Kafka的消息，那么在Kafka世界中，它就是一个消费者。
流是如何不同的，因为这也消耗或产生信息给卡夫卡？为什么需要它，因为我们可以使用consumer编写自己的消费者应用程序，并根据需要对它们进行处理，或者将它们从消费者应用程序发送到Spark？
我做了谷歌这个，但没有得到任何好的答案。如果这个问题太琐碎的话，很抱歉。
浏览 2提问于2017-05-17得票数 158
回答已采纳
我将构建一个基于Java的Kafka流应用程序，它将连续地侦听一个主题X，获取数据，执行一些基本的清理，并写入Oracle数据库。kafka集群不在我的域中，无法在其中部署任何代码或配置。
设计这样一个解决方案的最佳方法是什么？我遇到了Kafka流，但对于它是否可以用于“Topic> Process >Topic”场景感到困惑吗？
浏览 0提问于2018-11-27得票数 0
回答已采纳
3回答
当我从Kafka主题创建一个流并打印它的内容时
import os
os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.0.2 pyspark-shell'
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka impo
浏览 2提问于2019-12-08得票数 1
回答已采纳
1回答
我有一个用java编写的REST api，希望避免使用kafka对数据库或其他外部服务进行任何额外的查询。因此，如果客户端应用程序向/some/service发送请求，我将检查是否有一个名为"/some/ service“的主题，并且在该主题中是否已经存在一个日志，该日志对应于该服务在过去24小时内对当前用户的先前响应，否则我将只执行该服务并将返回的值存储在该主题中。
kafka的这种用法正确吗？而且，我知道kafka很快，但是有没有一种方法可以在不检查每个日志的情况下识别存储在主题中的所有响应中的哪些属于当前用户？
浏览 7提问于2018-08-16得票数 0
我有一个真正的环境与3 kafka机器集群，这是接收大量数据。对于每个主题，有25个复制因子设置为2的分区。 我的应用程序(基于kafka流的应用程序)从这个kafka集群获取数据，它停机了一个多月。现在，每个分区中都有大量的延迟；总共有90000000个。 我知道以下参数： max.poll.records ; default —> 500
max.partition.fetch.bytes ; default —> 1048576
fetch.max.bytes ; default —> 52428800
fetch.min.bytes ; default —>
浏览 25提问于2019-05-22得票数 1
我有一个使用apache的简单管道，我想使用现有的kafka puplisher处理器在kafka主题中发布一些消息。
问题是如何使用apache表达式语言指定kafka键？我厌倦了类似于${message:jsonPath('$.key')}的东西，但是，当然，我得到了一个错误，因为对象message不存在。
我还尝试使用filename对象，它类似于输入消息的默认对象名称，但没有帮助。
使用另一个kafka处理器可以通过设置message key field属性来实现，但是PublishKafka处理器呢？
浏览 0提问于2018-09-20得票数 1
回答已采纳
3回答
我在Flink中做实时流，其中Kafka是消息队列。我正在申请120秒的EventTimeSlidingWindow。滑动1秒。我还在事件时间的每一秒插入水印。
我担心的是，如果元素会在水印之后延迟出现，会发生什么？现在在我的情况下，Flink简单地丢弃了它各自的水印之后的消息。filnk是否提供了任何机制来处理这种延迟消息，比如维护单独的窗口？我也浏览了文档，但我没有弄清楚。
浏览 19提问于2017-01-19得票数 3
回答已采纳
2回答
我是新的火花和卡夫卡和我有一个稍微不同的使用模式火花流与卡夫卡。我在用
spark-core_2.10 - 2.1.1
spark-streaming_2.10 - 2.1.1
spark-streaming-kafka-0-10_2.10 - 2.0.0
kafka_2.10 - 0.10.1.1
连续事件数据正在被流到一个kafka主题中，我需要从多个火花流应用程序中处理这个主题。但是当我运行星火流应用程序时，只有其中一个接收数据。
Map<String, Object> kafkaParams = new HashMap<String, Object>(
浏览 3提问于2017-08-29得票数 4
回答已采纳
我们正面临着一个关键的决定，我们需要为一个实时遥测处理和监测系统选择合适的工具。
该系统有两个主要目的：
在仪表板UI应用程序中显示实时遥测数据(高刷新率)
处理来自多个源的大量数据
我们有两个问题：
在我们的用例中，大多数数据使用者将是web客户端，与我们目前发现的情况不同的是，-> -- 和 --大多用于基于云的平台，其中大多数数据使用者是服务而不是客户端。
通常，来自Kafka或RabbitMQ的数据使用者是负载均衡的，并且消息不是为每个客户端复制的。
。
卡夫卡或RabbitMQ是否适合为每个客户端复制数据的环境？
这种工具是否适合在为每个
浏览 0提问于2018-01-23得票数 5
1回答
火花流写卡夫卡与延迟后x分钟
、、、、
我们有一个火花流应用程序。体系结构如下
从火花到卡夫卡。
Spark应用程序正在使用qubole/ Kinesis进行来自Kinesis的结构化流。然后将数据进行聚合，然后推送给卡夫卡。
我们的用例需要延迟4分钟才能进入卡夫卡。
加窗时间为2分钟，水印为4分钟。
val windowedCountsDF = messageDS
.withWatermark("timestamp", "4 minutes")
.groupBy(window($"timestamp", "2 minutes", "2 minute
浏览 0提问于2019-06-06得票数 2
回答已采纳
1回答
在运行在多个实例中的基于Kafka的分布式JVM应用程序中，我需要对特定Kafka主题中的特定消息在特定可配置时间内(此超时值由业务逻辑驱动，可能会发生变化)的事件采取操作。
我如何才能以集群安全的方式实现这一点？
浏览 0提问于2020-09-29得票数 2
4回答
同步来自多个数据源的数据
、、、、
我们的团队正在尝试构建一个预测维护系统，其任务是查看一组事件，并预测这些事件是否描述了一组已知的异常。
我们正处于设计阶段，目前的系统设计如下：
这些事件可能发生在IoT系统的多个源(如云平台、边缘设备或任何中间平台)上。
事件由数据源推入消息队列系统(目前我们选择了Apache )。
每个数据源都有自己的队列(Kafka主题)。
从队列中，数据由多个推理机(实际上是神经网络)消耗。
根据特征集的不同，推理引擎将订阅多个Kafka主题和来自这些主题的流数据，以持续输出推理。
总体架构遵循单一责任原则，这意味着每个组件将彼此分离，并在单独的Docker容器中运行。
浏览 0提问于2019-05-27得票数 8
回答已采纳
分布式计算的本质是将执行与数据放在同一位置，或者换句话说，将代码发送到数据，而不是将数据发送到代码。这是Hadoop、Spark等的核心设计。
Kafka / Kafka Streams支持这样的设置吗？如果是，是如何实现的？如果没有，是否有计划，也许作为一个子项目，例如使用Kubernetes或类似的？
我知道我们可以为一个主题定义消费者组，但我不明白如何将分区分配给消费应用程序实例，以及这种分配是否可以支持位于同一位置的实例。
请告诉我是否有更好的词来搜索，因为"kafka consumer co-location“并没有取悦谷歌之神:/
浏览 0提问于2017-11-22得票数 1
目前，我们使用Flink FsStateBackend检查点并设置fileStateSizeThreshold，将写入HDFS上avro/json文件的数据大小限制在128 on。在检查点操作延迟一定时间后也关闭文件。
由于我们在新项目中没有使用高级Flink特性，所以我们希望使用Kafka Connect HDFS Connector中的Kafka流直接将消息写入hdfs (而不是旋转Flink)。
但是，我无法找到限制卡夫卡连接器中hdfs文件的文件大小的选项，除了flush.size，它似乎限制了记录的#。
如果连接器上没有设置，人们如何以另一种方式管理hdfs上流数据中的文件大小？
浏览 0提问于2018-08-08得票数 1
回答已采纳
1回答
我有一个简单的基于Kafka流的Spring应用程序，它使用来自传入主题的消息，进行map转换并打印这条消息。配置如下的KStream
@Bean
public KStream<?, ?> processingPipeline(StreamsBuilder builder, MyTransformer myTransformer,
PrintAction printAction, String topicName) {
KStream<String, JsonNode> source = builder.stream(topicName,
浏览 1提问于2019-05-28得票数 0
回答已采纳
1回答
我正在尝试运行一个Python脚本，该脚本在Spark1.3.1中运行得很好。我已经下载了Spark1.4并尝试运行这个脚本，但是它一直在说
星火流的卡夫卡库在类路径中找不到。尝试以下其中之一。
将Kafka库及其依赖项作为$bin/submit-packages org.apache.sight:streaming:1.4.0.
从Maven Central 下载工件的JAR，Group = org.apache.spark，artifact =，Version = 1.4.0。然后，将jar包含在星火提交命令中，如
.spark=‘spark 1’>.
我在提
浏览 2提问于2015-07-08得票数 3
我有点困惑，我想要澄清一下。我有点事要做。我想要有一个Kafka Streams拓扑，它将有五个独立的KStreams从各自的主题读取数据，并将数据转储到一个大型的整体主题中。接下来，我将创建一个GlobalKTable，它将从这个整体主题中读取内容，并实现一个全局存储，比方说，称为lookupStore。我想让这个物化的全局存储基本上作为其他Kafka Streams应用程序的“查找”表。我已经阅读了一些关于使用带有application.server配置的RPC层来公开这一点的文章，该RPC层将以一些独特的host:port的形式出现。 现在，我想要有许多独立的微服务，每个微服务都是Ka
浏览 41提问于2020-11-11得票数 0
我们在IBM上的Kafka中有一个潜在的问题，在运行/轮询过程中，我们的卡夫卡消费者(包括和KafkaStreams)偶尔会得到异常的ListenerConsumer
我们通过设置AuthorizationExceptionRetryInterval克服了卡夫卡听众中的这个问题，这意味着卡夫卡将重新尝试直到异常得到解决。这里一切都很好
[ consumer-0-C-1] o.a.k.c.consumer.internals.Fetcher : [Consumer clientId=consumerxxx_processing00b773de-064b-4c6f-8d0b-31a06
浏览 6提问于2021-11-02得票数 0
9.0客户端使用来自运行在远程system.My生产者上的两个代理的消息的客户端运行良好，能够向代理发送消息，但我的使用者无法使用这些messages.Consumer，而producer在我的本地系统上运行，两个代理在aws上运行。每当我试图运行消费者。下面的错误出现在代理日志上。
ERROR Closing socket for /122.172.17.81 because of error (kafka.network.Processor)
org.apache.kafka.common.protocol.types.SchemaException: Error reading fiel
浏览 10提问于2016-06-01得票数 5
回答已采纳
我正在做一个关于Kafka流和ktables的poc，我想知道是否有任何方法可以在kafka中存储数据(键-值对或键-对象对)，可以通过流、ktables、状态存储，这样我就可以根据键和值检索数据库。我创建了一个基于主题的kstream，在这个kstream上我推送了一些消息，并使用wordcountalgo在kstream上面创建的ktable中填充值。如下所示：
StoreBuilder customerStateStore = Stores.keyValueStoreBuilder(Stores.persistentKeyValueStore("customer-store
浏览 0提问于2019-07-24得票数 2
1回答
我有一些关于卡夫卡流以及它们是如何工作的问题，我对卡夫卡的消费者和生产者范式有着丰富的经验，然而，这是我第一次尝试使用卡夫卡流。
问题：
一般来说，卡夫卡消费者模型我们订阅了一个主题并开始从一个分区开始消费，为了简单起见，为了简单起见，我们有一个分区和一个消费者，现在如果我们想要增加我们的处理，我们增加了分区的数量，并且增加了更多的消费者，这在Kafka流中是如何工作的，如果我们增加分区，我们应该如何扩展应用程序，我们需要添加更多的机器还是我们需要做其他的事情？
当我通过Kafka消费者消费数据时，我可能最终会对消息做一些事情，例如，我可能会查询API下载文件并写入NFS并将消息转发或
浏览 1提问于2018-12-22得票数 1
回答已采纳
我在我的项目中使用了，需要将它集成到一个反应本地应用程序中，这也是项目的一部分。基本上，我的服务器是一个卡夫卡生产者和应用程序需要倾听主题和成为一个消费者。
一开始，我尝试使用这个npm包。。
但是这个包依赖于nodejs的核心模块，这些模块不是的一部分。有没有其他的方法让我可以在一个本地应用程序中使用Kafka进行交流呢？谢谢!
浏览 5提问于2017-03-01得票数 4
回答已采纳
我们是kafka的新手，我们有几个团队在一些应用程序上工作，这些应用程序相互之间发布/订阅事件。由于kafka主题名称将在团队之间共享，有没有命名的最佳实践？
基本上，我们不希望看到团队A命名一个主题companyname-appname-events，而团队B以完全不同的风格命名另一个主题productname_functionB。
如有任何建议，欢迎光临！
注意:这可能听起来更像下面提出的问题：然而，那里的作者问的是更具体的问题。
浏览 0提问于2017-05-02得票数 24
回答已采纳
3回答
我通过ffmpeg使用了hls流媒体。
在流开始时，会有几秒钟的延迟。
当我想要在非常延迟的时候从一个4k的视频流中抓取。
我能做些什么？
浏览 9提问于2017-04-23得票数 3
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
