Avro Kafka在scala和Python之间的转换问题

文章/答案/技术大牛

发布

3回答

为什么我会得到这样的编译错误：“找不到kstream.Consumed的隐含值”，我如何修复它？

scala、apache-kafka、avro、apache-kafka-streams、jsonschema

一个这样生成的case类有一个字段，就是其中一个值。在AVRO模式中，这是用type=t1，t2表示的，所以生成看起来很不错，这是一个sum类型:可以是t1类型，也可以是t2类型。问题变成了在从topic到case类(二进制-> Avro Map -> case类)的反序列化路径上缺少什么。因此，另一个依赖项是帮助AVRO GenericRecord到case类的映射和返回。我们还有一些手工编写的

浏览 22提问于2020-04-08得票数 2

回答已采纳

1回答

python、scala、apache-kafka、spark-avro

我们的项目同时包含scala和python代码，我们需要向kafka发送/消费avro编码的消息。val r

浏览 8提问于2017-02-11得票数 0

回答已采纳

1回答

Scala:读取火花结构化流中的Kafka Avro消息时出错

scala、apache-spark、apache-kafka、avro、spark-structured-streaming

我一直在尝试阅读Kafka的avro串行化消息，这些消息来自于Scala2.11的火花结构化流(2.4.4)。为此，我使用了星星之火-avro(下面的依赖性)。我使用合流卡夫卡库从python生成kafka消息。星火流可以使用模式来使用消息，但它不能正确读取字段的值。我准备了一个简单的例子来说明这个问题，这里的代码是可维护的：我在</e

浏览 0提问于2020-02-29得票数 2

回答已采纳

1回答

无法将催化剂类型IntegerType转换为Avro类型["null“、"int"]

apache-spark、pyspark、apache-kafka、avro

我已经用Pyspark构建了Spark结构化流流程，它从kafka主题读取avro消息，进行一些转换，并将数据作为avro加载到目标主题中。我使用ABRIS包()与Schema集成，从汇合式序列化/反序列化Avro。"null", ],}, 此过程引发以下错误：Cannot convert Catalyst type IntegerType to Avro我

浏览 0提问于2020-07-28得票数 3

1回答

用于火花流的org/apache/kafka/common/serialization/ByteArraySerializer :java.lang.NoClassDefFoundError

java、apache-spark、pyspark、apache-kafka、apache-spark-sql

我试图在Ubuntu上运行一个星火流应用程序，但是我遇到了一些错误。由于某种原因，Ubuntu22.04没有找到jar文件，尽管Windows上的配置是相同的。", 2) \此外，我还在中下载并定位了---- avro和sql的jar文件。sources.jarspark-avro_2.12-3.3.0-test-sources.jar 我的spark版本为3.3.0，scala</em

浏览 6提问于2022-08-15得票数 0

回答已采纳

1回答

通过kafka- Avro* -控制台-生成器生成avro序列化对象的问题*

apache-kafka、avro、oracle-golden-gate

我正在使用kafka-avro-console-producer二进制文件使用生成消息，方法是： at io.confluent.kafka.formatter.AvroMessageReader.readMessage(AvroMessageReader.java:180)at kafka.tools.ConsoleProducer$.main(ConsoleP

浏览 1提问于2020-02-25得票数 1

回答已采纳

1回答

Avro GenericData记录应该在哪里转换为实现Avro SpecificRecord的CustomEvent？

spring、apache-kafka、avro、spring-kafka、confluent-platform

使用spring-boot 2.4.4, spring-kafka 2.6.7, kafka client 2.6.0, confluent 6.6.1 (用于KafkaAvroSerializer和架构注册表)、Scala 13。production“中，会使用适当的CustomEvent @Payload来调用@KafkaListener，因此在某个地方会发生从字节到Avro GenericRecord再到CustomEventSpecificRecord的

浏览 37提问于2021-04-13得票数 0

1回答

星火流卡夫卡消费者(Avro) - AttributeError：'dict‘对象没有属性’拆分‘

python、apache-spark、pyspark、apache-kafka、avro

我正在尝试构建一个Streaming，它用Avro格式的消息从Kafka主题中消耗消息，但是我面临着汇合消息反序列化器的一些问题。按照的说明，我让Kafka使用者正确地反序列化消息，但最终未能运行PythonStreamingDirectKafkaWordCount示例。import KafkaUtils from confluent_kafka.avro.cached_schema_registry_client import Ca

浏览 0提问于2018-03-17得票数 0

2回答

如何通过Kafka* Connect自定义源连接器使用Avro生成的模式*

apache-kafka、avro、apache-kafka-connect

我正在使用Kafka Connect开发一个自定义源连接器，并且我正在尝试合并Avro支持。为此，我创建了一些.avsc文件来描述键和值模式，并将avro-maven-plugin添加到项目中，以自动创建可在代码中使用的相应java类。从生成的类中，我可以获得作为org.apache.avro.Schema类型的对象的模式。但是，源连接器的poll方法的返回类型是一组org.apache.k

浏览 18提问于2020-03-16得票数 0

回答已采纳

2回答

弹簧卡夫卡测试与合流卡夫卡阿夫罗串行化找不到ZKClientConfig

java、spring、spring-boot、avro、spring-kafka

我很难开始嵌入式卡夫卡的集成测试。显示此错误： <artifactId>kafka-avro-serializer</artifactId>

浏览 5提问于2020-06-19得票数 0

回答已采纳

2回答

在scala火花流中使用foreach时，不希望字符串作为类型？

scala、utf-8、deserialization、spark-streaming、avro

流接收到的avro编码消息。string在scala中默认使用UTF-16编码。或者一种让avroRecord本身进入utf-8的方法。被困在这里。需要为这个问题陈述一个前进的</e

浏览 5提问于2017-06-27得票数 1

回答已采纳

1回答

在火花流/结构化流中阅读Kafka的avro消息

pyspark、apache-kafka、spark-streaming、spark-structured-streaming、spark-streaming-kafka

火花版本: 2.3.0卡夫卡版本: 2.2.0 我有一个卡夫卡生产者，它以avro格式发送嵌套数据，我正在尝试用火花流/结构化流写代码，这将反序列化从kafka到dataframe的avro，做转换，把它以拼花格式写成我能够在spark/scala中找到avro转换器，但是还没有增加对pyspark的支持。如何在火花放电中转换相同的内容。谢谢。

浏览 3提问于2019-05-02得票数 3

回答已采纳

10回答

将Spark结构化流与合流模式注册表集成

apache-spark、apache-kafka、avro、confluent-schema-registry、spark-structured-streaming

我正在使用一个卡夫卡源在星火结构化流接收汇合编码的阿夫罗记录。我打算使用，但与火花结构化流的集成似乎是不可能的。我已经看到了这个问题，但无法让它与合流模式注册表一起工作。

浏览 0提问于2018-02-20得票数 28

回答已采纳

1回答

如何使用融合模式注册表与from_avro标准函数？

apache-spark、apache-kafka、avro、spark-structured-streaming、confluent-schema-registry

我的Kafka和Schema注册表基于合流社区平台5.2.2，我的Spark有2.4.4版本。我启动Spark env的时候： .option("subscribe", topicName)我得到了关于键和值的模式信息： import io.confluent.kafka.schemaregistry.client.rest.RestServicejava.util.concurrent.Thread

浏览 0提问于2019-09-16得票数 2

1回答

给avro和kafka* SourceRecord的拼花文件*

java、apache-kafka、parquet、avro、apache-kafka-connect

我正在尝试将拼花文件转换成SourceRecords。我目前正致力于从avro模式生成连接模式。static Record readRecord(ParquetReader<Record> reader) { }AvroParquetReader生成的av

浏览 11提问于2022-10-07得票数 0

回答已采纳

2回答

阅读Kafka中的Avro格式消息- Pyspark结构化流

apache-spark、pyspark、avro、spark-avro

我正在尝试使用PySpark 2.4.3阅读Kafka的Avro消息。基于下面的堆栈over flow链接，可以转换为Avro格式(to_avro)，并且代码正在按预期工作。但是，from_avro并没有工作，而且在issue.Are下面，还有其他模块支持阅读来自卡夫卡的avro消息吗？这是Cloudra的分布环境。请对此提出建议。.__/\_,_/_/ /_

浏览 1提问于2019-11-21得票数 1

1回答

来源连接器:有趣的字节字符

apache-kafka、mqtt、apache-kafka-connect、mosquitto

我遵循连接摩斯基托和卡夫卡与MQTT源连接器。我正在把摩斯基托出版商发送给摩斯基托用户和卡夫卡消费者的数据。但是我的卡夫卡-使用者的ConsumerRecord对象中的键和值字段有一些占优势的字节字符。下面是我得到的代码片段和输出。checksum=None，serialized_key_size=17，serialized_value_size=43，serialized_header_size=62) 是什么导致了卡夫卡消费者中额

浏览 0提问于2018-12-18得票数 3

回答已采纳

3回答

安装合流-kafka* avro与pip*

python、apache-kafka、avro、confluent-platform

我试图安装阿夫罗软件包的汇合-卡夫卡与python3上的macOs塞拉利昂。pip install confluent-kafka[avro]no matches found: confluent-kafka[avro] 如何安装这个软件包？

浏览 3提问于2017-11-14得票数 4

回答已采纳

2回答

如何从JSON对象生成Kafka模式

python、node.js、apache-kafka

我有一个包含不同值类型的嵌套键的示例JSON数据，例如整数、浮点数和字符串： "ordertime": 1497014222380, "itemidMountain View", "zipcode": 94041} 我需要编写一个模式，以便在中注册，这样就可以用JSON_SR、AVR

浏览 13提问于2022-08-04得票数 0

回答已采纳

1回答

如何在阅读卡夫卡的消息流的同时处理阿夫罗的信息？

apache-spark、spark-structured-streaming

下面的代码从Kafka读取消息，消息在Avro中，所以我如何解析该消息并将其放入Spark2.2.0中的数据文件中？Dataset<Row> df = sparkSession.readStream() .option("kafka.bootstrap.servers", "localhost:9092") .o

浏览 2提问于2017-12-16得票数 3

回答已采纳

点击加载更多

为什么我会得到这样的编译错误：“找不到kstream.Consumed的隐含值”，我如何修复它？