《kafka》专题

我想用Confluent Schema Registry和Avro Schema evolution测试Spring Cloud Stream，以将其与我的应用程序集成。我发现Spring Cloud Stream不支持到融合模式注册中心的安全连接，实现仍然非常基础。因此，我决定将Confluent Schema Registry Client与Spring Kafka一起用于模式注册表部分，其余

我使用结构化流媒体（Spark 2.0.2）来消费Kafka消息。使用scalapb，protobuf中的消息。我得到以下错误。请帮助。。 线程“main”scala中的异常。ScalaRefltionException：不是一个术语org.apache.spark.sql.catalyst.符号$SymbolApi$9.apply术语（Seflection.scala:592）org.apach

我正在尝试从Kafka读取JSON消息并将它们存储在具有火花结构化流的HDFS中。 我遵循了下面的示例，当我的代码如下所示时： 然后我得到hdfs中具有二进制值的行。 这些行按预期连续写入，但采用二进制格式。 我发现了这个帖子： https://databricks.com/blog/2017/04/26/processing-data-in-apache-kafka-with-structure

我正在使用Spark 2.2上的Spark结构化流媒体将文件从HDFS目录流式传输到Kafka主题。我想为我写的主题数据捕捉Kafka偏移量。 我正在使用 给Kafka写信。 当我利用 为了捕获流的进度信息，检索到的信息与Kafka中创建的偏移量不相关。 我假设这是因为流提供的信息实际上是关于我正在使用的文件流的，而与Kafka中编写的内容无关。 有没有一种Spark Structure流式处理方

我第一次使用pyspark。Spark版本：2.3.0Kafka版本：2.2.0 我有一个Kafka制作人，它以avro格式发送嵌套数据，我正试图在pyspark中编写spark流/结构化流的代码，它将来自Kafka的avro反序列化为数据帧，并进行转换，将其以拼花格式写入s3。我在spark/scala中找到了avro转换器，但pyspark中的支持尚未添加。如何在pyspark中转换相同的值。

我的想法是使用Spark流媒体Kafka从Kafka巴士上获取事件。在检索了一批avro编码的事件之后，我想用Spark avro将它们转换为SparkSQL数据帧，然后将数据帧写入配置单元表。 这种方法可行吗？我是Spark的新手，我不完全确定是否可以使用Spark Avro包来解码Kafka Events，因为在留档中只提到了avro文件。但到目前为止，我的理解是，这是可能的。 下一个问题是：

我正在使用火花流做分析。分析后，我必须将kafka消息保存在hdfs中。每个kafka消息都是一个xml文件。我不能使用，因为它会保存整个rdd。rdd的每个元素都是kafka消息（xml文件）。如何使用火花在hdfs中保存每个rdd元素（文件）。

我有以下Flume代理配置来读取来自kafka源的消息并将它们写回HDFS接收器 如果每个轮询周期只有一条kafka消息到达，则kafka消息内容是avro数据，并且正确地序列化为文件。 当两个kafka消息到达同一批次时，它们被分组在同一个HDFS文件上，因为avro消息包含两个模式数据，结果文件包含模式数据模式数据，导致它是无效的. avro文件。 如何拆分avro事件以将不同的kafka消息

下面是Kafka的spark streaming代码。在这里，我试图获取批处理的密钥作为Dstream，然后将其转换为列表。以便对其进行迭代，并将与每个键相关的数据放入以该键命名的hdfs文件夹中。 关键基本上是-模式。表\u名称 正在提取密钥，但其类型为DStream[字符串] 将其转换为列表并更新var final\u list\u of\u键 现在尝试遍历列表。 但我遇到了一个错误-不支持在

我正在努力定制我的spring kafka streams应用程序。我一直试图在我的KStreams上配置处理未捕获（运行时异常）。 参考文档https://docs.spring.io/spring-cloud-stream-binder-kafka/docs/3.1.0/reference/html/spring-cloud-stream-binder-kafka.html#_kafka_st

我有严重的问题处理Spring云流Kafka活页夹。Spring Cloud 3.0.2.Release的配置设置中存在许多模糊性和一致性问题。我一直试图为Kafka主题设置组ID和客户端ID，但是尽管尝试了各种不同的组合，我还是无法正确配置组ID。 文档声称，我们应该能够通过配置以下设置之一来设置组id和客户端id:https://cloud.spring.io/spring-cloud-sta

从Kafka Streams 2.5.0开始，拓扑似乎必须包含一个输入主题。在Kafka2.4.1（以及更早的版本）中，情况并非如此。 我有一个应用程序，其中的拓扑只是创建一些全局状态存储，从其他应用程序写入的主题中读取数据。 使用Kafka 2.5.0，我得到以下错误： 如果添加一个虚拟输入主题（例如，通过)，应用程序启动良好。 这种行为是意料之中的，还是Kafka Streams 2.5.0中

配置的application.yaml如下所示。这个想法是我有3个输入和3个输出主题。该组件从input topic获取输入，并将输出提供给OutputTopic。 引发的异常为 谁能帮助我与Kafka Streams Spring-Kafka代码样本处理与多个输入和输出主题。 更新：2021年1月21日

我有一个Spring-boot应用程序，可以听Kafka。为了避免重复处理，我尝试手动提交。为此，我在阅读主题后异步提交了一条消息。但是我被困在如何实现消费者幂等，这样记录就不会被处理两次。

2.LocalDateTime字段上的序列化程序注释 为了确保我有正确的对象映射器设置和必要的依赖关系，我创建了一个rest控制器，将响应模拟为json作为restendpoint返回一个带有日期时间字段的对象，这将正确返回；示例：