如何确定在kafka producer中使用哪种序列化？
在确定在Kafka Producer中使用哪种序列化时，需要考虑以下几个因素：
- 数据格式：首先，需要确定要发送的数据的格式。如果数据是结构化的，例如JSON、XML或Avro等，可以选择相应的序列化器。如果数据是非结构化的，例如文本或二进制数据，可以选择字节序列化器。
- 性能要求：不同的序列化器在性能方面可能会有差异。一些序列化器可能更高效，可以提供更好的吞吐量和延迟。因此，如果对性能有较高的要求，可以选择性能较好的序列化器。
- 兼容性：如果已经有现有的消费者应用程序，需要确保选择的序列化器与消费者应用程序兼容。如果消费者应用程序使用特定的序列化器进行反序列化，那么生产者应该使用相同的序列化器。
- 生态系统支持：考虑选择的序列化器在Kafka生态系统中的支持程度。一些序列化器可能有更广泛的社区支持和文档资源，可以更容易地找到解决方案和支持。
根据上述因素，可以选择以下常见的序列化器：
- JSON序列化器：适用于结构化的JSON数据。可以使用Kafka提供的JsonSerializer或第三方库如Jackson进行序列化。腾讯云相关产品：云消息队列 CMQ，产品介绍链接：https://cloud.tencent.com/product/cmq
- Avro序列化器：适用于结构化数据，提供了强大的架构演化和兼容性支持。可以使用Confluent提供的AvroSerializer或其他Avro库进行序列化。腾讯云相关产品：消息队列 CKafka，产品介绍链接：https://cloud.tencent.com/product/ckafka
- Protobuf序列化器：适用于高性能和紧凑的二进制数据序列化。可以使用Google的Protobuf库进行序列化。腾讯云相关产品：云原生数据库 TDSQL-C，产品介绍链接：https://cloud.tencent.com/product/tdsqlc
- 字节序列化器：适用于非结构化的文本或二进制数据。可以使用Kafka提供的ByteArraySerializer进行序列化。
需要注意的是，选择序列化器时应根据具体需求进行评估和测试，以确保选择的序列化器能够满足性能和兼容性要求。
相关·内容
在阅读kafka文档时，我在kafka producer中发现了许多类型的序列化，但我对如何确定任何用例的序列化感到困惑。
浏览 18提问于2016-09-23得票数 0
问题是，producer在multiprocessing.Process进程中工作时不会产生消息。('KAFKA_HOST')KAFKA_NAME = os.getenv('KAFKA_NAME')
producer =KafkaProducer(bootstrap_servers=f'{KAFKA
浏览 3提问于2022-02-11得票数 0
我需要更改Kafka设置中的值序列化程序/反序列化程序(出于测试目的，我一直在使用IntegerSerializer/IntegerSerializer)。使用JAVA API，它可以完全按照预期工作；但是，在使用控制台工具时，它似乎不能正常工作。
我所做的所有故障排除都让我得出了一个结论:kafka控制台生产者似乎忽略了序列化选项。我尝试了--property value.serializer、--pro
浏览 20提问于2017-06-28得票数 5
回答已采纳
3回答
我正在尝试使用Spring Cloud Stream来集成Kafka。正在编写的消息是Java POJO，虽然它的工作方式与预期一样(消息正在写入主题，我可以使用消费者应用程序读取)，但消息的开头添加了一些未知字符，这在尝试集成Kafka Connect以接收主题中的消息时会造成麻烦，所以我希望能够在application.yml属性文件中执行此操作，而不是通过代码执行此操作。然而，当yml被更新以指定序列化程序时，它并没有像我预期的那样工作，即它没有生成与在Java中
浏览 1提问于2017-11-21得票数 3
回答已采纳
我正在尝试理解如何在Spring Cloud Stream中使用不同的键序列化器/反序列化器。我已经找到了指定全局serdes的方法，但我不知道如何为每个绑定指定不同的serdes，以便允许不同的键类型(Integer，String等)。例如，以下配置使用属性spring.kafka.consumer.keyDeserializer和spring.kafka.producer.keySerializer指定全局键serdes
浏览 20提问于2019-05-09得票数 1
回答已采纳
我在获取KafkaProducer实例化来接受avro序列化程序的值序列化程序配置属性时遇到了问题。然后，我将以下内容添加到我的项目POM文件中的依赖项中： <groupId>io.confluent</groupId> <version>5.2.2</ve
浏览 43提问于2019-11-12得票数 0
如何使用kafka-console-producer在kafka topic中插入json文件数据？每个json数据集可以存储为一条消息吗？last_name": "Lindt", "gender": "Male",}
如
浏览 2提问于2019-02-13得票数 11
我计划使用kafaka在pyspark中发送数据。通过搜索资料，我发现我需要对producer进行自定义序列化，以便可以广播该对象。但是我如何在pyspark中实现这个功能呢？spark==2.2.1public class KafkaProducer implements Serializable {
浏览 43提问于2019-08-14得票数 0
我正在使用Kafka 0.8测试版，我只是试图处理发送不同的对象，使用我自己的编码器序列化它们，并将它们发送到现有的代理配置中。就目前而言，我正试图让DefaultEncoder工作。我的制片代码是：import kafka.producer.KeyedMessage;
import kafka.produce
浏览 2提问于2013-09-26得票数 5
回答已采纳
1)在Java Client中，我们可以使用producer.put()为整数和字符串添加序列化程序。我们如何为confluent-kafka-python客户端做到这一点?2)我们如何序列化protoBuf记录？
浏览 12提问于2017-03-02得票数 0
1回答
在此流处理I事件后，我在Kafka KTable changelog中看到以下值： {
"topic": "events-consumer-v1-assessments-store-changelog"我预计如果我将null作为值推送到聚合操作中，它将被删除，但看起来像是留下了一些垃圾，我不确定它的映射器问题或不正确的删除方式或正确的KTable行为。
浏览 22提问于2019-01-25得票数 0
回答已采纳
1回答
我在structs中定义了一组Thrift，如下所示： 1: i32 a,}(a)将Foo实例序列化为与Binary兼容的字节(使用Binary协议或Compact协议)如何将Thrift序列化实例发送到Kafka集群？
提前感谢
浏览 3提问于2017-03-18得票数 3
回答已采纳
另外，我需要一些特殊的序列化逻辑。据我所知，要创建自定义序列化程序，我必须实现Serializer<>接口。因此，问题是Kafka-镜制造者没有使用我在producer.properties中指定的自定义序列化程序：client.id=producerB要构建序列化程序，我使用gradle： constraints {
com
浏览 1提问于2019-07-18得票数 2
回答已采纳
即: Flink 1.8 -> (宁静，正式支持Druid0.9.2) -> Druid 0.20.1
未来: Flink 1.11 -> Kafka -> Druid 0.20.1使用推荐的.Flink应用程序生成Scala类的DataStream (在下面的代码中为ExampleData)。根据，Scala类属于“Flink提供的特殊序列化器”，但我不确定它是如何被序列化的，因此当Druid从Kafka主题读取它时，它是如何</e
浏览 5提问于2021-08-06得票数 0
回答已采纳
我正在编写一个消费者，它监听Kafka主题，并在消息可用时消费消息。我通过在本地运行Kafka来测试逻辑/代码，它工作得很好。<String, GenericRecord> producer = configureProducer();
at org.apache.kafka.cli
浏览 0提问于2019-08-20得票数 3
我已经建立了汇流数据平台，并开始开发一个SourceConnector，并在相应的SourceTask.poll()方法中执行以下操作(下面是伪Java代码)： records.add(new SourceRecord(sourcePartition, sourceOffset, topic, org.apache.kafka.connect.data.Schema.BYTES_SCHEMA我希望使用Schema，以便
浏览 5提问于2016-05-19得票数 2
回答已采纳
2回答
我在Python2.7中使用Avro Producer。这是我的守则： kafkaBrokers = os.environ.get('KAFKA_BROKERS') topic = os.environ.get('KAFKA_TOPIC')
subject = '${}-valu
浏览 6提问于2020-08-25得票数 0
回答已采纳
11回答
在库更新之前的Spring / Kafka应用程序中，我使用以下类org.telegram.telegrambots.api.objects.Update将消息发送到Kafka主题。现在我使用以下org.telegram.telegrambots.meta.api.objects.Update。如你所见-他们有不同的包裹。=${kafka.host}:${kafka.port}
spring.kafka.consumer.auto-offse
浏览 2提问于2018-08-04得票数 36
回答已采纳
1回答
我有一个Kafka实例正在运行(在本地，在一个Docker中)，我使用在Go中创建了一个生产者。由于我想在我的主题中使用Kafka流，所以生产者必须在消息中嵌入一个时间戳，否则我会得到一条丑陋的错误消息：
org.apache.kafka.streams.errors.StreamsException:输入记录ConsumerRecord(主题= crawler_events，分区= 0，偏移量= 0，CreateTime = -1，序列化<
浏览 0提问于2018-10-08得票数 4
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
