我正在将我的Spark Scala App Kafka API升级到版本0.10。我过去常常为消息的反序列化创建自定义方法，该方法以字节串格式提供。然而，我找不到任何关于如何创建自定义Avro模式反序列化器的信息，以便我的kafkaStream可以在我createDirectStream和使用来自Kafka的数据时使用它。
有可
浏览 2提问于2017-07-10得票数 3
回答已采纳
msg = kafka_consumer.poll(timeout=2.0)
显然，Avro编码值是二进制的，而不是Unicode。使用Java API，我可以为KafkaConsumer构造函数指定一个KafkaAvroDeserializer。Python API似乎没有与此类似的配置选项。可以传递给底层librdkafka的配置选项似乎没有指定任何自定义(反)序列化程序或解码器或avro相关选
浏览 2提问于2016-07-07得票数 0
我配置了一个kafka连接器(postgres到topic)，我想用火花流消费者来阅读它。./kafka-avro-console-consumer --bootstrap-server localhost:9092 --property schema.registry.url=http://得到这一结果：
null {"id":1557，"ip":{"string":"90.2
浏览 0提问于2019-04-09得票数 2
回答已采纳
我有一个spark 2.0应用程序，它使用spark streaming从kafka读取消息(使用spark-streaming-kafka-0-10_2.11)。结构化流媒体看起来真的很酷，所以我想试着迁移代码，但我不知道如何使用它。
在常规的流中，我对createDstrean使用kafkaUtils，并且在我传递的参数中，它是值反序列化器。在结构化流中，文档说我应该使用Dat
浏览 3提问于2016-11-20得票数 9
如何使用直播流API指定kafka spark流的消费组id。使用spark1.3
kafkaParams.put("group.id", "app1");
浏览 0提问于2016-04-09得票数 7
回答已采纳
我正在尝试使用Confluent InfluxDB Sink Connector将来自kafka主题的数据放入我的InfluxDB中。首先，我使用nifi将日志文件中的数据传输到kafka topic，效果良好。Connect UI创建了InfluxDB接收器连接器，我得到了以下异常： org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask(WorkerSinkTask.java:
浏览 37提问于2019-03-19得票数 0
回答已采纳
我正试图用Kafka 0.10构建一个POC。我使用自己的scala域类作为Kafka消息，其中包含一组字符串数据类型。我不能使用Kafka库附带的默认序列化类或字符串序列化器类。我想我需要编写自己的序列化程序并将其提供给生产者属性。如果您知道用Kafka (使用scala)编写一个示例自定义
浏览 1提问于2017-04-17得票数 0
我正在尝试调试为什么不可能反序列化我们通过Kafka接收的数据，并基于一些特定的模式使用avro序列化。
它看起来像这样。有一个用java编写的生产者，它根据某种avro模式序列化数据，将数据发送到Kafka主题，而在消费者端有一个用Golang编写的程序。我在Kafka中使用了avro库和。Unmarshaler: kafka</em
浏览 1提问于2021-09-12得票数 1
1回答
我正在尝试听一个话题，这个话题是我使用confluent的kafka连接功能发布的。但是，我无法对其进行反序列化。我相信它的avro序列化，但不能找到正确的反序列化。{"c1":{"int":10},"c2":{"string":"foo"},"create_ts":1552598863000,"update_ts":1552598863000}
下
浏览 1提问于2019-03-15得票数 1
在这一点上，我想使用“香草”卡夫卡部署，没有任何汇合插件。
这个是可能的吗？到目前为止，我发现的所有示例都很快开始使用针对avro消息的汇合特定工具。我确信，应该有一种方式，我可以发布和消费avro消息的只是卡夫卡平台，任何“发行特定”没有加载项。
浏览 3提问于2016-05-18得票数 6
回答已采纳
1回答
) \我的spark版本为3.3.0，scala
浏览 6提问于2022-08-15得票数 0
回答已采纳
下面的代码从Kafka读取消息，消息在Avro中，所以我如何解析该消息并将其放入Spark2.2.0中的数据文件中？Dataset<Row> df = sparkSession.readStream() .option("kafka.bootstrap.servers
浏览 2提问于2017-12-16得票数 3
回答已采纳
我正在尝试构建一个Streaming，它用Avro格式的消息从Kafka主题中消耗消息，但是我面临着汇合消息反序列化器的一些问题。按照的说明，我让Kafka使用者正确地反序列化消息，但最终未能运行PythonStreamingDirectKafkaWordCount示例。/bin/spark-submit --packages org.apache.spark:spark-streaming-<e
浏览 0提问于2018-03-17得票数 0
默认情况下，当使用上述代码时，avroRecord是一个字符串。string在scala中默认使用UTF-16编码。import org.apache.spark.streaming.api.java.JavaStreamingContext
import org.apache.spark.streaming.kafka源代码编译为/home/spark_sca
浏览 5提问于2017-06-27得票数 1
回答已采纳
10回答
我正在使用一个卡夫卡源在星火结构化流接收汇合编码的阿夫罗记录。我打算使用，但与火花结构化流的集成似乎是不可能的。
我已经看到了这个问题，但无法让它与合流模式注册表一起工作。
浏览 0提问于2018-02-20得票数 28
回答已采纳
1) kafka集群-版本为0.10 2)星团- 1.6，其中有0.9卡夫卡罐运行火花提交作业时，火花选择0.9版本的卡夫卡。以下是我们的观察
1) Producer - works精细( 0.9 api和0.10 api生产者兼容) 2)使用KafkaUtils的卡夫卡消费者流- works精细(这里似乎也有0.9API和0.10api生产者兼容) 3
浏览 3提问于2016-10-24得票数 1
回答已采纳
1回答
kafka stream / Ksql是否实际上以某种方式支持json？支持的其他格式是什么？我已经看到，是否有可能把扁平的json解释为桌子。我想更好地理解这一部分；通过Ksql可以通过SQL查询kafka-streams的其他格式是什么？这是怎么可能的或支持的？本地的支持是什么？
浏览 0提问于2019-07-02得票数 2
回答已采纳
1回答
我使用汇合JDBC连接器连接到postgres数据库，以检索更改并将它们放在kafka主题中。现在，我想使用spring引导使用者来使用这些消息。这些消息都是AVRO格式的。我从连接器中获得了模式，并使用avro maven插件为它生成了一个POJO类。当我不使用avro反序列化数据时，我将接收数据，但无法读取。= http:/
浏览 1提问于2021-06-18得票数 1
回答已采纳
我没有看到如何使用camel-avro组件来产生和使用kafka avro消息的示例？目前我的骆驼路线是这样的。为了使用camel-kafka-avro consumer & producer来使用模式注册表和其他类似的道具，应该对其进行哪些更改。Server -> Log ");
from("kafka:{{consumer.t
浏览 1提问于2019-03-16得票数 3
2回答
我正在使用Kafka jars在scala shell上执行spark代码，我的目的是从Kafka主题流式传输消息。我的spark对象已经创建好了，但是有没有人可以帮助我在启动spark shell时如何传递jaas配置文件呢？我的错误将我指向缺少jaas配置
浏览 1提问于2018-09-05得票数 4
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
云直播活动推荐
腾讯云开发者
