问题标签 [confluent-schema-registry]

在接收到 dstream 后，在使用 kafka 和模式注册表的 spark 流中，如何将 dstream 批处理转换为 spark 中的 Dataframe？

从 confluent 使用 KafkaAvroDecoder 后的 Dstream 类型是 Dstream(String,Object)。当我使用下面的代码时，它会在 avro 列中将架构数据类型（例如 Int 更改为 Long）。

代码参考

Object.toSting 和读取为 json 会丢失 int 的架构。除了在数据框列中转换类型之外，还有其他方法吗？

etc/schema-registry/schema-registry.properties

kafka 和 zookeeper 已经在运行。

为什么动物园管理员的日志不断出现

我想知道这可能是模式注册表失败的原因。

有什么建议么。

我在 node.js 中实现了 Avro 模式，模式与消息有效负载一起发送。它工作正常。我正在寻找是否有任何方法可以将模式注册表与 Kafka-node 模块一起使用。我已经探索过，但没有成功找到任何东西。

并在每条消息中发送模式会增加消息大小？与使用模式注册表相比，它会影响性能吗？

对此的任何帮助将不胜感激。

我使用 Confluent Platform 3.3 作为 Kafka 连接器，同时使用以下命令启动连接器，

收到以下错误

Zookeeper 在 kafka 客户端 2181 端口中运行，我尝试通过以下命令启动模式

但它没有显示任何错误消息，但端口 8081 没有启动。请帮我解决这个问题。

我正在尝试使用 Avro 序列化程序和模式注册表将对象发送到 Kafka。
这是一个简化的代码：

我假设架构是从注册表中“在幕后”读取的，并且对象（用户）是序列化的，但我得到下面的错误。
我错过了什么？
我是否必须明确阅读架构并发送 GenericRecord？

org.apache.kafka.common.errors.SerializationException：序列化 Avro 消息
时出错 原因：java.lang.IllegalArgumentException：不支持的 Avro 类型。支持的类型有 null、Boolean、Integer、Long、Float、Double、String、byte[] 和 IndexedRecord
at io.confluent.kafka.serializers.AbstractKafkaAvroSerDe.getSchema(AbstractKafkaAvroSerDe.java:123) ~[kafka-avro-serializer-3.3 .0.jar!/:?]
at io.confluent.kafka.serializers.AbstractKafkaAvroSerializer.serializeImpl(AbstractKafkaAvroSerializer.java:73) ~[kafka-avro-serializer-3.3.0.jar!/:?]
at io.confluent .kafka.serializers.KafkaAvroSerializer.serialize(KafkaAvroSerializer.java:53) ~[kafka-avro-serializer-3.3.0.jar!/:?]
在 org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:424) ~[kafka-clients-0.9.0.1.jar!/:?]

如果我们在 Kafka 中使用模式注册表，是否每个生产者每次将记录发送到代理时都需要发送当前版本的 Kafka？

如果是，那么这个额外开销的含义是什么，因为我们已经在每个 avro 文件中发送了模式？

如果不是，请对我的问题感到愚蠢，并请帮助我更好地理解模式注册表。

我在单独的三个 centOS 机器上创建了一个 kafka 集群。所有 Zookeeper 和 kafka 服务器都运行正常。

当我尝试运行模式注册表时，主机 Zookeeper 已关闭，因此模式注册表引发了错误，例如错误服务器意外死亡。

供您参考，我已经配置了 kafka 商店 url，就像zookeeper.connect在 kafka 服务器配置中一样，用逗号分隔。但没有运气。有什么想法吗？

我正在使用具有查询模式的 jdbc 源连接器，似乎没有指定的表名，在模式注册表中为记录键和记录值注册的模式具有空模式名称，并且被分配了默认名称“ConnectDefault”如 Confluent 的 AvroData 类中定义的那样https://github.com/confluentinc/schema-registry/blob/master/avro-converter/src/main/java/io/confluent/connect/avro/AvroData.java

使用生成的 avro 源和 SpecificAvroSerde 运行 Kafka Streams 应用程序时，出现错误：

我试图从主题中发布一个新版本的键和值模式，并使用表名作为模式名，并删除\"name\":\"ConnectDefault\",\"namespace\":\"io.confluent.connect.avro\"没有运气的属性的原始版本。我是否缺少名为 ConnectDefault 的类，或者我可以在源连接器的某处指定没有命名空间的架构名称？

我的卡夫卡流配置：

我的 Kafka 连接配置：

我有一个带有第一个架构的 avro 文件，然后我更新了附加到同一文件的架构。所以现在我在一个文件中有两个模式。avro 如何处理这种情况。我会在文件中添加任何新字段还是在读取这些数据时丢失任何数据。这是一个实时流应用程序，我将数据写入 hdfs。我的上游系统可能会更新架构，但 hdfs 编写器可能在旧架构上。所以 hdfs avro 文件将有两个模式，直到我更新编写器以处理更新的模式。

注意 - 我没有架构注册表，我每天创建一个 avro 文件。因此，如果在一天中更新了一个模式，我将拥有一个包含两个模式的 avro 文件。

我按照这个文件，它运作良好。现在我尝试使用 spark 中的连接器数据。有什么可以参考的吗？由于我使用的是 confluent，它与原始 kafka 参考文档有很大不同。

这是我到目前为止使用的一些代码。问题是无法将记录数据转换为 java.String。（并且不确定这是正确的消费方式）