《kafka》专题
-
Spring cloud Stream：具有单独jaas配置的Kafka生产者和消费者的多个绑定器不能一起工作
-
在spring cloud stream和kafka 1.0.1中使用Kstream
我试图创建一个简单的程序来打印一个Kafka主题的Kstream。我不断地得到一个NPE和完全没有想法。 我已经使用了spring cloud-stream-binder-kafka-streams依赖项，并且我正在使用spring cloud的最新版本“Finchley.m9”。 我写的代码是： Application.Properties具有： 当我启动服务时，我在控制台上不断得到以下错误：
-
spring-cloud-stream kafka消费者并发
使用Spring-Cloud-Stream的kafka绑定器，如何配置并发消息消费者（在单个消费者jvm中）？如果我没有理解错的话，在使用kafka时并发使用消息需要分区，但是s-c-s文档指出，要使用分区，您需要通过partitionKeyExpression或PartitionKeyExtractorClass在生成器中指定分区选择。Kafka博士提到循环分区。 s-c-s文档根本没有提到sp
-
在所有Kafka消费者结束后完成Flink程序
我在这里设置了一个最小的示例，其中有N个Kakfa主题的N个流（在下面的示例中为100个）。 我想在每个流看到“EndofStream”消息时完成它。当所有流都完成时，我希望Flink程序能够顺利完成 当parallelism设置为1时，这是正确的，但通常不会发生。 从另一个问题来看，似乎并非Kafka消费群体的所有线索都结束了。 其他人建议抛出异常。但是，程序将在第一个异常时终止，并且不会等待所
-
如何模拟Kafka记录元数据
尝试使用Mockito为下面的代码编写单元测试，但我遇到了错误使用方法的异常。结果不为空，已验证，也进入循环。 这是我所写的： 例外情况：
-
Kafka:=线程“main”组织中的异常。阿帕奇。火花sql。流动。StreamingQueryException:找不到连接的条目
同时运行Kafka代码 1）错误流执行：查询[id=c6426655-446f-4306-91ba-d78e68e05c15， runId=420382c1-8558-45a1-b26d-f6299044fa04]终止与错误java.lang.ExceptionIn初始azerError 2） 线程“针对[id=c6426655-446f-4306-91ba-d78e68e05c15，runId=
-
Kafka最早和最新偏移值之间的区别是什么
发送消息1、2、3、4 接收消息1、2、3、4 崩溃/断开连接 发送消息5、6、7 对于这种结果，我必须使用哪个值，以及其他更改/配置需要做什么
-
Java Kafka结构化流
批处理查询中似乎不支持“最新”。我想知道是否有可能用另一种方法做类似的事情（不直接处理偏移）
-
Spring Kafka流处理器不工作
我正在尝试使用Spring boot编写一个Kafka流处理器，但当消息产生到主题中时，它不会被调用。 主题消息有不同的类型，并且是Avro格式的。在模式注册表中使用Avro UNION注册模式。 这些是主题 application.yml我正在使用cp-all-in-one-community作为docker-file 但现在我得到以下错误：
-
如何为Kafka KeyValueStore设置值解/序列化程序？
我将来自Kafka主题的消息存储在KeyValueStore中，以便以后可以查询它们。我创建了一个KTable，如下所示： 我在application.yml中配置了使用者，如下所示： 但是，当我从KeyValueStore读取时，键会正确地作为字符串返回，但返回的值是字节数组，而不是MyMessage。由于某种原因，我的自定义反序列化程序未被使用。我尝试自己反序列化消息，但我的反序列化器发生了异
-
Spring boot Kafka类反序列化-不在受信任包中
并将这些行添加到 但仍会出现以下错误： org.apache.kafka.common.errors.SerializationException：反序列化偏移量1处分区topic2-0的键/值时出错。如有需要，请通过记录继续使用。原因：java.lang.IllegalArgumentException：类“com.SpringMiddleware.Entities.Crime”不在受信任的包[
-
数据丢失Spark 2.1-kafka broker 0.8.2.1流媒体
1流媒体和Kafka broker版本0.8.2.1，我在AWS上为spark和Kafka提供了单独的服务器。 使用直接进近，我希望从流媒体中获得30个字符串，但实际接收范围只有15-25个。交叉检查Kafka消费者在300秒内显示30个字符串。还有小溪。foreachRDD{rdd= 获取最终数据背后有什么问题。我正在使用火花会话创建sc和ssc。 谢谢你。
-
Apache Kafka和Spark流媒体
我正在读这篇博文： http://blog.jaceklaskowski.pl/2015/07/20/real-time-data-processing-using-apache-kafka-and-spark-streaming.html 它讨论了如何使用Spark Streaming和Apache Kafka进行一些近实时处理。我完全理解这篇文章。它确实展示了我如何使用Spark Stream
-
从Kafka读取时如何异步制作火花流
我有一个Kafka分区，和一个parkStreaming应用程序。一个服务器有10个内核。当火花流从Kafka收到一条消息时，后续过程将需要5秒钟（这是我的代码）。所以我发现火花流读取Kafka消息很慢，我猜当火花读出一条消息时，它会等到消息被处理，所以读取和处理是同步的。我想知道我可以异步读取火花吗？这样从Kafka读取的数据就不会被后续处理拖动。然后火花会很快消耗来自Kafka的数据。然后我可
-
如何制作Spark流媒体（Spark 1.0.0）从Kafka（Kafka Broker 0.8.1）读取最新数据
我的火花流应用程序从Kafka获取数据并对其进行处理。 如果应用程序失败，大量数据存储在Kafka中，并且在Spark Streaming应用程序的下一次启动时，它会崩溃，因为一次消耗了太多数据。由于我的应用程序不关心过去的数据，因此只消耗当前（最新）数据完全没关系。 我找到了“auto.reset.offest”选项，它在Spark中的行为几乎没有什么不同。如果配置了zookeeper，它会删除