在spark structured streaming中反序列化kafka avro主题时，int编码无效
在Spark Structured Streaming中反序列化Kafka Avro主题时，如果int编码无效，可能是由于以下原因导致的：
- 数据类型不匹配：检查Avro主题中的字段类型是否与Spark Structured Streaming中的字段类型匹配。如果Avro主题中的字段类型为int，而Spark Structured Streaming中的字段类型为其他类型（如long），则会导致反序列化失败。
- 编码格式不正确：确保Avro主题中的int字段使用正确的编码格式进行序列化。常见的编码格式有"int"、"long"、"fixed"等，需要根据实际情况选择正确的编码格式。
- Avro模式不匹配：检查Avro主题中的模式定义是否与Spark Structured Streaming中的模式定义匹配。如果模式定义不匹配，可能会导致反序列化失败。
解决该问题的方法包括：
- 检查数据类型匹配：确保Avro主题中的字段类型与Spark Structured Streaming中的字段类型一致。如果不一致，可以通过转换或映射来解决类型不匹配的问题。
- 检查编码格式：确认Avro主题中的int字段使用正确的编码格式进行序列化。可以参考Avro官方文档或相关资料了解不同编码格式的使用方法。
- 检查模式定义：确保Avro主题中的模式定义与Spark Structured Streaming中的模式定义一致。如果不一致，可以通过修改模式定义来解决反序列化失败的问题。
对于推荐的腾讯云相关产品和产品介绍链接地址，由于要求不能提及具体的云计算品牌商，建议参考腾讯云的相关文档和产品介绍页面，查找与Spark Structured Streaming、Kafka Avro主题反序列化等相关的产品和解决方案。
相关·内容
2回答
在spark structured streaming中反序列化kafka avro主题时，int编码无效
apache-spark、apache-spark-sql、byte、spark-structured-streaming
我正在尝试使用spark structured streaming (2.3.1版)处理来自kafka的avro数据流，所以我尝试使用this示例来反序列化。仅当主题value部件包含StringType时才有效，但在我的示例中，架构包含long and integers，如下所示： public static final String USER_SCHEMA = "{"
+ "\"type\":\"record\","
+ "\"name\":\"variable
浏览 56提问于2019-04-30得票数 1
1回答
如何在运行时在DStream中找到值的模式？
apache-spark、apache-kafka、spark-streaming、apache-spark-1.6
我使用的是Spark 1.6和Kafka 0.8.2.1。
我正在尝试使用Spark Streaming从Kafka获取一些数据，并对这些数据进行一些操作。
为此，我应该知道获取的数据的模式，有什么方法可以做到这一点，或者我们可以通过使用字段名从流中获取值吗？
浏览 0提问于2017-05-29得票数 1
1回答
Scala:读取火花结构化流中的Kafka Avro消息时出错
scala、apache-spark、apache-kafka、avro、spark-structured-streaming
我一直在尝试阅读Kafka的avro串行化消息，这些消息来自于Scala2.11的火花结构化流(2.4.4)。为此，我使用了星星之火-avro(下面的依赖性)。我使用合流卡夫卡库从python生成kafka消息。星火流可以使用模式来使用消息，但它不能正确读取字段的值。我准备了一个简单的例子来说明这个问题，这里的代码是可维护的：
我在python中创建记录，记录的模式是：
{
"type": "record",
"namespace": "example",
"name": "RawR
浏览 0提问于2020-02-29得票数 2
回答已采纳
1回答
可以在没有HDFS的情况下使用Spark Structured吗？
spark-structured-streaming
我经常使用HDFS和Kafka，我注意到Kafka比HDFS更可靠。所以现在使用Spark-structured-streaming，我很惊讶检查点只有HDFS。用Kafka来勾点会更快更可靠。那么，有没有可能在没有HDFS的情况下使用spark structured？奇怪的是，我们只能在Kafka中将HDFS用于流数据。或者，有没有可能告诉斯帕克忘记ChekpPointing并在程序中管理它？ Spark 2.4.7 谢谢
浏览 39提问于2021-05-07得票数 2
1回答
用于kafka avro解码器消息的spark数据集编码器
apache-kafka、spark-streaming、avro
导入spark.implicits._ val ds1 = spark .readStream .format("kafka") .option("kafka.bootstrap.servers"，"localhost:9092") .option("subscribe"，"test") .option("startingOffsets"，"latest") .load() .asKafkaMessage .select($“value”。as[ArrayByte]) .map(ms
浏览 1提问于2017-11-06得票数 0
1回答
在星火读取序列化输入中不可序列化的任务
java、serialization、apache-spark、apache-kafka、avro
我正在研究一个基于星火的卡夫卡消费者，它以Avro格式读取数据。下面是try代码读取和处理输入。
import java.util.*;
import java.io.*;
import com.twitter.bijection.Injection;
import com.twitter.bijection.avro.GenericAvroCodecs;
import kafka.serializer.StringDecoder;
import kafka.serializer.DefaultDecoder;
import scala.Tuple2;
import org.apache.
浏览 2提问于2016-03-15得票数 0
1回答
使用pyspark从apache kafka反序列化avro时的空列
apache-spark、pyspark、apache-kafka、avro、confluent-schema-registry
我正在用Kafka，Spark和jupyter笔记本做概念验证，我遇到了一个奇怪的问题。我正在尝试读取Avro记录，从kafka到pyspark。我使用confluent模式注册表来获取模式来反序列化avro消息。在对spark dataframe中的avro消息进行反序列化后，结果列为空，没有任何错误。该列应该包含数据，因为当转换为字符串时，某些avro字段是可读的。
我也试过在Scala的spark-shell上做这个(没有jupyter)，我已经尝试了一个基于docker的spark以及一个独立的spark安装
我按照这个SO主题获得了from_avro和to_avro函数：
jars
浏览 1提问于2019-06-25得票数 0
1回答
GCP Dataproc -在提交无法工作的作业时添加多个包(kafka，mongodb)
apache-spark、google-cloud-platform、dependency-management、spark-structured-streaming、dataproc
我正在尝试在提交dataproc pyspark作业时添加kafka & mongoDB包，但是这是失败的。到目前为止，我只使用了kafka包，这很好，但是当我尝试在下面的命令中添加mongoDB包时，会出现错误
命令正常工作，只使用Kafka包。
gcloud dataproc jobs submit pyspark main.py \
--cluster versa-structured-stream \
--properties spark.jars.packages=org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.2, s
浏览 9提问于2022-05-24得票数 0
回答已采纳
4回答
使用Spark 2.0.2读取来自Kafka的Avro消息(结构化流)
scala、apache-kafka、spark-streaming、avro、apache-spark-2.0
我有一个spark 2.0应用程序，它使用spark streaming从kafka读取消息(使用spark-streaming-kafka-0-10_2.11)。
结构化流媒体看起来真的很酷，所以我想试着迁移代码，但我不知道如何使用它。
在常规的流中，我对createDstrean使用kafkaUtils，并且在我传递的参数中，它是值反序列化器。
在结构化流中，文档说我应该使用DataFrame函数进行反序列化，但是我不能确切地理解这是什么意思。
我看过像这个这样的例子，但是我在Kafka中的Avro对象非常复杂，不能像例子中的字符串那样简单地强制转换。
到目前为止，我尝试了这种代码(我在这
浏览 3提问于2016-11-20得票数 9
1回答
在hadoop 3.2.0上构建spark2.4.3失败
apache-spark、hadoop、bigdata
我正在构建spark 2.4.3，使其与最新的hadoop 3.2.0兼容。
源代码可从下载
生成命令为./build/mvn -Pyarn -Phadoop-3.2 -Dhadoop.version=3.2.0 -DskipTests clean package
构建结果为：
[INFO] Spark Project Parent POM ........................... SUCCESS [ 1.761 s]
[INFO] Spark Project Tags ................................. SUCCESS [ 1.221 s]
浏览 41提问于2019-07-31得票数 3
回答已采纳
1回答
由于ClassCastException，闪烁流接收器ParquetWriter失败
java、apache-kafka、apache-flink、parquet
我已经编写了一个消费者，它读取卡夫卡主题，并使用StreamSink以拼图格式写入数据。但是我得到了以下错误
java.lang.Exception: org.apache.flink.streaming.runtime.tasks.ExceptionInChainedOperatorException: Could not forward element to next operator
at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.checkThrowS
浏览 6提问于2020-02-29得票数 0
1回答
找不到用于通过Kafka流式传输mongo db数据的数据集中存储的类型的编码器
scala、apache-spark、apache-kafka、spark-structured-streaming
我想跟踪Mongo oplog并通过Kafka进行流式传输。所以我找到了debezium Kafka CDC连接器，它尾随Mongo oplog的内置序列化技术。
模式注册表使用下面的转换器进行序列化，
'key.converter=io.confluent.connect.avro.AvroConverter' and
'value.converter=io.confluent.connect.avro.AvroConverter'
下面是我在项目中使用的库依赖项
libraryDependencies += "io.confluent" %
浏览 32提问于2018-08-10得票数 0
回答已采纳
4回答
来自Kafka的Spark流，并以Avro格式写入HDFS
scala、apache-spark、hdfs、spark-streaming、avro
我基本上是想使用Kafka中的数据，并将其写入HDFS。但是发生的情况是，它没有在hdfs中写入任何文件。它会创建空文件。
也请指导我，如果我想在hdfs中写avro格式，我可以如何修改代码。
为了简单起见，我写入本地C驱动器。
import org.apache.spark.SparkConf
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkContext
import org.apache.spark.streaming.Seconds
import o
浏览 4提问于2017-10-08得票数 0
1回答
如何为Apache SpecificRecordBase的所有子类选择一个编码器？
apache-spark、avro、apache-beam、serializable、encoder
背景
我的束流管道设计用于加工阿夫罗SpecificRecordBase型元件。
为了简化我的问题，假设我有两种以Avro格式生成的元素，它们都有自己的字段：
class Dog extends SpecificRecordBase {
....
}
class Cat extends SpecificRecordBase {
...
}
管道将读取输入Kafka中的元素，处理元素，并将处理后的元素放入输出Kafka中，如下所示：
Pipeline pipeline = Pipeline.create(getOptions());
pipeline.getCoderRegist
浏览 14提问于2020-03-02得票数 0
回答已采纳
1回答
如何在Spring中使用Avro序列化
java、apache-kafka、avro
我正在学习Kafka，现在是Avro，为了保持发送方对象和接收方对象之间的一致性，我们保留了一个JSON (.avsc)。但我找不到任何简单的例子来说明如何使用它。有些例子是使用confluent (是Avro的汇合任务)，有些是通过Avro工具生成对象。到目前为止，我有一个卡夫卡的工作设置。
对象类
package com.example.kafka;
public class Hello {
String name;
String age;
public Hello(String name, String age) {
this.name = name;
this.
浏览 0提问于2019-08-18得票数 1
2回答
与AVRO相比，在kafka流中使用JSON序列化程序是否会导致性能下降？
json、apache-kafka、apache-kafka-streams
我在Kafka Streams应用程序中使用了以下JSON配置，
properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG,
Serdes.String().getClass());
properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,
GenericJSONSerde.class);
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
浏览 3提问于2019-10-25得票数 3
回答已采纳
2回答
我们可以通过相同的kafka主题来引导不同类型的信息吗？
apache-kafka、apache-spark-sql、spark-streaming
我有一个场景，我有不同类型的消息从kafka producer中流式传输。
如果我不想针对不同的消息类型使用不同的主题，如何在spark-structured streaming消费者端处理它？
i.e. only one topic I want to use for different type of messages ...say Student record , Customer record....etc.
How to identify which message is been received from Kafka topic?
Please let me know how
浏览 10提问于2019-07-25得票数 1
2回答
如何在Spark Structured Streaming中读取Kafka和查询外部存储，如Cassandra？
apache-spark、apache-spark-sql
如何在Spark Structured Streaming中读取Kafka和查询外部存储，如Cassandra？
我从Kafka获得消息流，我想对它应用Map操作，对于每个键，我想查询像Cassandra这样的数据存储，并获得该键的更多信息，并在流上应用进一步的操作。我如何使用Spark Structured Streaming 2.2.0来做到这一点？
浏览 2提问于2017-09-07得票数 1
2回答
已启用DStream检查点，但DStreams及其函数不能序列化。
scala、spark-streaming
我想把DStream发送给卡夫卡，但它仍然不起作用。
searchWordCountsDStream.foreachRDD(rdd =>
rdd.foreachPartition(
partitionOfRecords =>
{
val props = new HashMap[String, Object]()
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, outbroker)
props.put(ProducerConfig.VALUE_SERIALIZER_CLAS
浏览 1提问于2016-10-31得票数 2
1回答
Kafka Connect:如何使用HDFS接收器将Kafka主题中的protobuf数据发送到hdfs？
apache-kafka、apache-kafka-connect
我有一个生产者，它产生一个主题的protobuf消息。我有一个消费者应用程序来反序列化protobuf消息。但是hdfs接收器连接器直接从Kafka主题获取消息。etc/schema-registry/connect-avro-standalone.properties中的键和值转换器将被设置为什么？做这件事最好的方法是什么？提前感谢！
浏览 2提问于2016-11-24得票数 5
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
即时通信 IM
对象存储活动推荐
腾讯云开发者
