avro 相关问题

Confluent Maven 存储库不工作？

需要用到Confluent kafka-avro-serializer Maven神器。从官方指南我应该将这个存储库添加到我的 Maven pom 融合 需要用到Confluentkafka-avro-serializerMaven神器。从 official guide 我应该将这个存储库添加到我的 Maven pom <repository> <id>confluent</id> <url>http://packages.confluent.io/maven/</url> </repository> 问题是 URL http://packages.confluent.io/maven/ 目前似乎不起作用，因为我得到以下回复 <Error> <Code>NoSuchKey</Code> <Message>The specified key does not exist.</Message> <Key>maven/</Key> <RequestId>15E287D11E5D4DFA</RequestId> <HostId> QVr9lCF0y3SrQoa1Z0jDWtmxD3eJz1gAEdivauojVJ+Bexb2gB6JsMpnXc+JjF95i082hgSLJSM= </HostId> </Error> 其实Maven并没有找到神器 <dependency> <groupId>io.confluent</groupId> <artifactId>kafka-avro-serializer</artifactId> <version>3.1.1</version> </dependency> 你知道问题是什么吗？ 谢谢 需要在 pom.xml 中添加 confluent 存储库 请在 pom.xml 中添加以下行 <repositories> <repository> <id>confluent</id> <url>https://packages.confluent.io/maven/</url> </repository> </repositories> 该文件可用，因为您可以直接访问它来下载它： http://packages.confluent.io/maven/io/confluent/kafka-avro-serializer/3.1.1/kafka-avro-serializer-3.1.1.jar 您可以尝试在 Maven 命令中添加 -U 标志以强制下载缓存文件。 repo 的根目录不可浏览，这就是为什么您在浏览到 http://packages.confluent.io/maven/ 时收到消息的原因 就像你一样，我使用公司存储库（Sonatype Nexus）并且无法代理 confluent 的存储库。 然后我更改了我的 maven settings.xml 以从镜像存储库中排除融合： <mirrors> <mirror> <id>nexus</id> <mirrorOf>*,!confluent</mirrorOf> <!-- mirror anything but confluent as Nexus cannot proxy it --> <url>repository.company.local/nexus/content/groups/public</url> </mirror> </mirrors> ... <repositories> ... <repository> <id>confluent</id> <url>http://packages.confluent.io/maven/</url> </repository> </repositories> 这样，工件解析也适用于合流的工件。 不像代理 repo 那样简洁，但至少比手动下载和注册每个依赖项更简单。 seams jar 文件从 http url 中删除或 http url 不工作。 https 网址对我有用。 <repositories> <repository> <id>confluent</id> <url>https://packages.confluent.io/maven/</url> </repository> 您可以在 Maven 设置文件中添加一个镜像，以从 confluent repo 中获取 jars 以及 repository config 。 需要改变的是 在settings.xml中添加镜像 <mirror> <id>confluent</id> <mirrorOf>confluent</mirrorOf> <name>Nexus public mirror</name> <url>http://packages.confluent.io/maven/</url> </mirror> 在 Maven 设置的存储库部分添加这个 <repository> <id>confluent</id> <url>http://packages.confluent.io/maven/</url> <releases> <enabled>true</enabled> </releases> <snapshots> <enabled>true</enabled> </snapshots> </repository> 尝试将 Artifactory 连接到 Confluent Maven 存储库时，您必须将 Artifactory 中的 repo URL 设置为 http://packages.confluent.io/maven 或 https://packages.confluent.io/maven（这两种方案似乎都可以正常工作）。令人困惑的部分是，当您要求 Artifactory 测试该 URL 时，它将失败并显示消息“输入可能不为空”。您也无法浏览 Artifactory 中的存储库。然而，不管这些问题如何，当客户端请求它们时，工件将被下载和缓存。 在 Maven 项目的 POM 文件中添加融合存储库。 在 SBT 项目的 build.sbt 中使用下面 resolvers += "confluent" at "https://packages.confluent.io/maven/" 有问题： <repositories> <repository> <id>confluent</id> <url>http://packages.confluent.io/maven/</url> </repository> </repositories> 工作： <repositories> <repository> <id>confluent</id> <url>https://packages.confluent.io/maven/</url> </repository> </repositories> 将“http”更改为“https”对我有用，因为 maven 出于安全原因阻止了 HTTP。

我使用什么工具来序列化/反序列化存储在 Kafka 主题中的 Avro 消息，其模式已使用 Pyspark 在模式注册表中注册？

我从 MySql 数据库中捕获了一些更改数据，使用 Avro 转换器将其序列化并将其存储在 Kafka 主题中。现在我正在尝试从 Kafka 读回 Avro 序列化数据。我可以……

Apache Kafka Avro 反序列化：无法反序列化或解码特定类型的消息。

我正在尝试将 Avro Serialize 与 Apache kafka 一起用于序列化/反序列化消息。我正在创建一个生产者，用于序列化特定类型的消息并将其发送到队列。当我...

Python：将 Avro 文件反序列化为 DF

我有一个 avro 文件输出，看起来像这样，我想从“身体”获取信息，但我无法成功地做到这一点。 { “序列号”：123， “偏移量”：“123”，...

Pubsub avro 模式测试：“消息对模式无效”

我正在尝试在 pubsub 中创建一个模式，我计划在创建主题时使用它。 我的 avro 模式是嵌套的，它看起来像这样： “类型”：“记录”， “名字”：...

AvroIO.writeGenericRecords(schema) 是否能够附加到 java 中的现有文件？

我一直在为这个问题绞尽脑汁，我正在使用 java 和 Avro，用窗口写入、分片和分片模板“W-P-sssss-of-nnnnn”编写文件，但现在

反序列化 Kafka Avro 记录 [重复]

我正在编写一个从 Apache Kafka 数据流读取的 Java 程序。这些消息是 Avro 序列化的。每条消息都包含一条 Avro 序列化记录。包 io.confluent.kafka.seria...

Kafka 提供商的 Avro Json

对于以下代码，我在行中出错 value_serializer=lambda m：io.DatumWriter(avro_schema).write(m).bytes() 说 TypeError: write() 缺少 1 个必需的位置参数：'encoder' 什么...

每次迭代递归返回 json 输出 python

我是 python 的新手，我试图创建读取 json 的 avro，在此之前我试图通过 csv.DictReader() 在 python 中使用 csv 生成 json，在这里我想返回一个函数，它的基因...

Flink Avro 期望类型是 PojoTypeInfo

我有 avro 类的实例，我想编写一个自定义文件接收器来处理 avro DataStream。 val list = List.fill(1000) { Person.newBuilder() // avro 类 .setName("...

如何从 python 类创建 Avro 模式？

如何将如下所示的简单 python 类转换为 avro 模式？ 类德图（SQLModel）： 名称：海峡 测量值：整数 这是 Testo.schema() 输出 { “标题”：“...

使用 Python 中的 AVRO，仅用于序列化

我想使用 Avro 对跨不同应用程序的对象序列化的二进制格式进行标准化。应用程序本身不会使用 Avro RPC 进行通信。可能是什么...

Avro 模式演变（向后兼容）使用 pyspark 结构化流返回 null

Avro 向后兼容返回空记录。 我将一条由 schema_ver1.avsc 编码的记录和一条由 schema_ver2.avsc 编码的记录发送到 kafka， 然后，我查询名为

使用嵌套数组从无头 avro 创建数据框

我正在尝试读取一个具有外部模式文件的无头 avro 文件。无头 avro 有一个结构类型的嵌套数组。我正在尝试将它读取到 spark 数据帧并在 parsi 中面临问题...

无法使用 pyspark readstream 从 kafka 主题读取记录数组

我正在使用来自 kafka 主题的 pyspark readstream 以及一系列记录，例如 [ {}, {}, {} ]。 我能够使用 from_avro( F.col('value'), avro_schema ) 解析单个记录。 然而，实际...

SpringBoot @KafkaListener 得到 MessageConversionException：无法从 A 转换为 B

我正在使用 SpringBoot 2.7.5 遇到了一个问题，@KafkaListener 收到了 MessageConversionException。整个错误日志如下所示： Bean [example.package.api.kafka.HelloKafkaListener@30e312a2]; ...

双重自由或腐败 C++ Avro

我需要在 C++ 中使用 Avro 序列化一个 JSON 字符串。 我安装了 libserdes 库（https://github.com/confluentinc/libserdes）并使用了示例代码 ./examples/kafka-serdes-avro-console-

如何将可选的类列表定义为 Avro Schema 中的字段之一以避免 Avro 序列化错误？

我的要求是将数据存储在数据库中：如果地址是通过邮递员作为请求发送的，则不应发送 currentAddress 和 previousAddresses，如果两者都已发送，则地址

Apache Avro 别名与模式注册表

我正在尝试使用别名重命名文件名并从 avro 模式中生成 java 类。使用新名称生成消息，但我的消费者应用程序在 s 中具有旧名称或别名...

如何让 Spark avro 阅读器在读取分区时停止推断类型

尝试从 hdfs 读取按年月日分区的 avro 文件。 我期望分区被读取为字符串，但类型被推断为整数 所以说我正在阅读 avro 文件......