使用kafka实现架构注册表时出错
使用Kafka实现架构注册表时出错可能是由于以下原因导致的：
- 配置错误：确保你的Kafka集群的配置正确，包括正确的主机名、端口号、认证信息等。可以通过检查配置文件或者命令行参数来确认配置正确性。
- 主题不存在：在使用Kafka作为注册表时，需要确保所使用的主题已经存在。如果主题不存在，可以使用Kafka的命令行工具创建主题或者在代码中动态创建主题。
- 权限问题：如果你的应用程序没有足够的权限访问Kafka集群，那么会出现出错的情况。确保你的应用程序拥有足够的权限，或者联系管理员进行权限配置。
- 应用程序代码问题：检查你的应用程序代码，确保使用Kafka的API正确，并且错误处理逻辑完备。例如，确保你的应用程序能够正确处理连接错误、消息发送失败等情况。
- 网络问题：如果你的应用程序无法连接到Kafka集群，可能是由于网络问题导致的。检查网络配置，确保能够与Kafka集群正常通信。
针对架构注册表的实现，Kafka是一个高吞吐量的分布式消息队列，可以用于实现可靠的数据传输和消息处理。架构注册表是用于存储和管理服务架构的元数据和配置信息的系统。
优势：
- 高吞吐量：Kafka能够处理大量的消息并提供高吞吐量的消息传输能力，适用于需要处理大量实时数据的场景。
- 可扩展性：Kafka的分布式架构使得它能够方便地扩展以应对不断增长的数据流量需求。
- 可靠性：Kafka提供了消息的持久化和复制机制，确保消息在传输过程中不会丢失，并且能够进行故障恢复。
- 弹性和容错性：Kafka的分区和副本机制能够保证系统的弹性和容错性，即使某个节点发生故障，整个系统仍能正常运行。
应用场景：
- 日志收集和处理：Kafka可以用于收集和处理大量的日志数据，支持实时处理和离线分析。
- 流式处理：Kafka可以与流处理框架（如Apache Flink、Apache Spark）集成，用于实时处理和分析数据流。
- 异步通信：Kafka可以作为不同系统之间的异步通信中间件，实现解耦和高可靠性。
- 事件驱动架构：Kafka可以作为事件驱动架构的核心组件，实现事件的发布和订阅。
腾讯云相关产品推荐：
- 腾讯云消息队列 CMQ（Cloud Message Queue）：基于消息中间件的高可扩展性、高可靠性的消息队列产品，可用于构建分布式系统、微服务架构等。 链接：https://cloud.tencent.com/product/cmq
- 腾讯云云服务器 CVM（Cloud Virtual Machine）：提供高性能、可靠、安全的云服务器，可用于部署和运行Kafka集群。 链接：https://cloud.tencent.com/product/cvm
- 腾讯云流数据分析（Data Streaming Analytics，DSA）：可实时处理海量数据的流计算和数据分析服务，与Kafka集成，支持流式ETL、实时计算、数据分析等场景。 链接：https://cloud.tencent.com/product/dsa
请注意，以上是仅供参考的答案，具体的解决方案可能因实际情况而有所不同。
相关·内容
在我的Spark streaming工作中，我试图从Kafka主题中读取Confluentavro消息，并获得“在记录解析中检测到错误的记录”。
我试着调试了很多次，但还是找不出错误的记录。需要帮助理解如何从格式错误的行中获取记录。有没有一种方法可以打印avro消息，以查看消息有什么问题。
我的代码：
object AvroReadMessage extends App {
val spark = SparkSession.builder.master("local[*]").appName("AvroReadMessage")
.getOrCreate
浏览 0提问于2019-12-16得票数 0
1回答
我试图使用函数编程(和spring云流)从输入主题转换输入AVRO消息，并在输出主题上发布新消息。这是我的转换函数：
@Bean
public Function<KStream<String, Data>, KStream<String, Double>> evenNumberSquareProcessor() {
return kStream -> kStream.transform(() -> new CustomProcessor(STORE_NAME), STORE_NAME);
}
CustomProcessor是一个实现“转换
浏览 5提问于2021-04-06得票数 1
我有一个定义如下的消费者。它读取主题外的avro消息，并构造一个聚合数据的状态存储，该数据也是avro类型的。
@Bean
public Consumer<KStream<String, InputEvent>> avroTest() {
Serde<OutputEvent> serdeOutEvent = new SpecificAvroSerde<>(schemaRegistryClient);
return st -> st.groupByKey().aggregate(OutputEv
浏览 23提问于2021-02-27得票数 1
回答已采纳
我试图在同一个kafka和动物管理员集群上挂载两个kafka模式注册表实例。但酋长一家混在一起了。在运行这两个注册中心时，如果我使用api "kafka- schema -registry“注册一个模式，它似乎是在”schema-registry other“创建的，而不是像预期的那样显示在”kafka-schema-registry“中。我的配置是：
version: '2.1'
services:
zoo1:
image: zookeeper:3.4.9
restart: unless-stopped
hostname: zoo1
浏览 6提问于2019-10-25得票数 1
回答已采纳
我已经设置了aws MSK，我正在尝试从MSK到弹性搜索的记录。我能够将数据推入MSK格式为json格式。我想沉溺于弹性搜索。我能做好所有的正确设置。这就是我对EC2实例所做的
wget /usr/local http://packages.confluent.io/archive/3.1/confluent-oss-3.1.2-2.11.tar.gz -P ~/Downloads/
tar -zxvf ~/Downloads/confluent-oss-3.1.2-2.11.tar.gz -C ~/Downloads/
sudo mv ~/Downloads/confluent-3.1.2
浏览 1提问于2019-12-29得票数 1
回答已采纳
1回答
我们使用Kafka消息的无注册Avro模式，所以我们有我们的自定义Avro转换器，因为我们没有可用的模式注册表
问题是，当我们使用新元素升级模式时，旧模式不是向后或向前兼容的
浏览 1提问于2021-05-31得票数 0
1回答
让我们假设我们有一个设置如下。
模式演化兼容性设置为向后。
轮询从DB写入到Kafka topic.HDFS Sink连接器的数据，读取来自Kafka主题的消息，并以Avro格式写入HDFS。
就像我所理解的那样。
查询DB并从ResultSet.V1的JDBC元数据中生成模式col1，col2，col3.SchemaV1在模式注册表中注册。
源连接器轮询来自DB的数据，并将消息写入V1模式中的Kafka主题。
(问题1)当HDFS连接器从主题读取消息时，它是否根据模式注册表中的V1模式验证消息？
下一个DB架构将被更改。从表中删除列"col3“。
下次J
浏览 0提问于2019-07-17得票数 3
回答已采纳
1回答
我们有一个kafka集群，它的Avro模式存储在Confluent的模式注册表中。在最近重新部署(我们的流应用程序之一)时，我们开始在单个主题(EmailSent)上看到不兼容的模式错误。这是唯一失败的主题，每当新的EmailSent事件提交到该主题时，我们都会收到错误。
Caused by:org.apache.kafka.common.errors.SerializationException: Error registering Avro schema: {"type":"record","name":"EmailSent"
浏览 0提问于2019-01-25得票数 5
1回答
我正在尝试使用HDFS卡夫卡连接器将原始消息从kafka发送到HDFS。连接器配置如下所示
{
"name": "hdfs3-connector-test",
"config": {
"connector.class": "io.confluent.connect.hdfs3.Hdfs3SinkConnector",
"tasks.max": "1",
"topics": "test-topic&
浏览 6提问于2021-07-23得票数 1
回答已采纳
我已经创建了一个NiFi流，它最终将json记录发布为带有Avro编码值和字符串键的记录，使用模式中的模式。这是 in NiFi。
我现在试图使用Kafka (connect-standalone)使用JdbcSinkConnector将消息移动到PostgreSQL数据库，但得到了以下错误：
我已经确认我的为1的架构。
Worker Config:
bootstrap.servers=localhost:29092
key.converter.schemas.enable=false
key.converter=org.apache.kafka.connect.storage.StringC
浏览 0提问于2018-12-11得票数 0
回答已采纳
我用洗净的jdbc-接收器将我的数据从kafka加载到oracle。
但是我用数据在值上编写我的模式。
我不想用数据来写模式，我怎么才能写关于kafka主题的模式，然后我只想从我的客户端发送数据呢？
提前感谢
json数据
{
"schema": {
"type": "struct",
"fields": [
{
"field": 'ID',
"type":
浏览 2提问于2020-04-15得票数 4
回答已采纳
我们已经在生产卡夫卡了。不幸的是，这是一个旧版本，0.10.2。我想开始使用Confluent注册表，从社区版的汇合平台。这将意味着安装旧的3.2.2模式注册表映像，以便与我们的旧kafka兼容。
根据我在文档中所读到的，合流模式注册表似乎使用Kafka作为存储它的状态的后端。但是，生产/阅读Kafka主题的客户端独立于Kafka向模式注册中心进行对话。
因此，我想知道在生产中是否更容易管理，在一个容器中运行SchemaRegistry/ Kafka /，独立于我们的主要Kafka集群。然后我可以使用最新版本的一切。另一个好处是，站起来这个新的服务组件不会给现有的Kafka集群带来任何意想不到
浏览 3提问于2019-10-31得票数 0
回答已采纳
我使用以下命令启动了Kafka
docker run -p 2181:2181 -p 9092:9092 -p 8081:8081 --env ADVERTISED_HOST=`docker-machine ip \`docker-machine active\`` --env ADVERTISED_PORT=9092 spotify/kafka
现在，我编写了一个简单的程序，将一个字符串发送到一个kafka主题中。它工作起来没有任何问题。
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.99.100:9092&#
浏览 0提问于2016-05-16得票数 1
我正在尝试使用以下命令启动kafka-connect：
connect-distributed /etc/schema-registry/connect-avro-distributed.properties
，但是我得到了：
[2017-02-15 12:45:35,962] INFO Instantiated task mysql-adventureworks-source-0 with version 3.1.2 of type io.confluent.connect.jdbc.source.JdbcSourceTask (org.apache.kafka.connect.runtim
浏览 18提问于2017-02-15得票数 1
我制作了一个坞-撰写文件来设置Kafka和模式注册表。但是，我无法检查模式注册表集群是否运行良好。
如何检查架构注册表群集模式？
我的坞-撰写文件作为集群运行良好吗？我不确定它的效果是否好。我只能看到两个卡夫卡和两个注册表在运行。
这是我的拼图文件。
version: '2'
services:
zookeeper:
image: zookeeper:3.7.0
container_name: zookeeper-ri
ports:
- "2181:2181"
kafka1:
image: confluenti
浏览 10提问于2022-06-14得票数 0
我正在实现一个生成kafka消息的过程，并且每个消息都应该有schema Registry验证的模式。为了进行开发，我使用docker运行kafka和Schema Registry，我的schema由模式注册表ui注册。
看起来我的模式没有被验证，或者我缺少一些配置。我的生产者类有以下代码：
package br.com.xx.realtime_transformation.producers;
import io.confluent.kafka.serializers.AbstractKafkaAvroSerDeConfig;
import io.confluent.kafka.seri
浏览 0提问于2020-04-28得票数 2
我试图使用Postgres接收器连接器将json消息加载到Postgres数据库中。我一直在线阅读，并且只找到了在JSON消息中使用模式的选项，但是，理想情况下，我不希望在消息中包含模式。有没有一种方法可以在模式注册表中注册JSON模式并像使用Avro那样使用它？
另外，我目前正在通过下载bin运行kafka，因为由于ARM兼容性问题，我在运行kafka与坞连接时遇到了几个问题。模式注册表是否有类似的安装？因为我只能选择通过汇流下载它并在码头上运行它。是否可以只运行带有docker的架构注册表，保持当前的设置？
谢谢
浏览 2提问于2022-01-31得票数 0
到目前为止，我所得到的唯一答案是，您必须给模式和主题以相同的名称，然后它应该将它们连接在一起。但是在注册了一个名为test_topic的模式之后，如下所示：
{
"type": "record",
"name": "test_topic",
"namespace": "com.test",
"doc": "My test schema",
"fields": [
{
"name": "na
浏览 1提问于2020-01-05得票数 1
回答已采纳
我想从弗林克州立大学出来和卡夫卡汇合。在中，为了进行模式检查并将数据放入到kafka主题中，我们所需要做的就是使用kafka客户端ProducerRecord对象和avro对象。
但是在中，我们需要重写卡夫卡出口的"ProducerRecord<byte[]，byte[]>序列化“方法。这将导致以下错误。
Caused by: org.apache.kafka.common.errors.SerializationException: Error registering Avro schema: "bytes"
模式登记和国家-卡夫卡出口似乎是不相容的。有什
浏览 6提问于2020-06-22得票数 0
回答已采纳
在我的消费者中，我想反序列化Kafka协议消息。键是字符串类型，但消息值是一个protobuf对象。我知道我必须为消息值创建自己的自定义反序列化程序，但不知道如何创建一个。下面是我的消费者实现，我需要替换标记的行：
using Confluent.Kafka;
using System;
using System.Threading;
namespace EventHubsForKafkaSample
{
class Worker1
{
public static void Consumer(string brokerList, string connStr,
浏览 44提问于2021-09-07得票数 1
我有一个生产者使用kafka connect，它使用Confluent Kafka Connect API，它以"SourceRecord“格式发布消息，其中包含”模式“和”结构“，如下所示。
我正在寻找一个示例代码来在scala中构建一个kafka消费者，它使用消息并将其反序列化为对象
import org.apache.kafka.connect.source.SourceRecord;
import org.apache.kafka.connect.source.SourceTask;
//publish kafka message in avro format
p
浏览 0提问于2018-08-24得票数 0
1回答
我是kafka的新手，.Here是我对不断变化的kafka模式的疑问。
我们如何在kafka消费者端处理模式更改？
如果我们在kafka publisher端更改负载结构，我如何确保kafka消费者端不会中断？
我很有兴趣知道全行业的实践来处理这种情况。
我不会为avro使用Confluent的模式注册表。还有没有其他经过试验和测试的选项？
浏览 0提问于2020-04-26得票数 0
我需要记录到达Kafka使用者的消息，因为对于Kafka消费者，没有Avro模式兼容性。
我怎样才能做到这一点？
我想，默认情况下，Avro不兼容的消息永远不会到达我的Kafka消费者代码。我使用Spring的@KafkaListener。
浏览 11提问于2021-12-07得票数 0
在我们分发给工人的道具中，我们有：
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=<dev-schema-url>
我们已经注意到，当我们使用模式注册中心时，性能是不好的。如果没有模式注册中心，我可以使用AvroConverter吗？我试过，但我们遇到了：
java.io.IOException Not a data file
模式注册表是否与kafka连接器读取的每一条消息相联系？
模式缓存了吗？
在没有模式注册中心的情况下使用A
浏览 1提问于2019-01-23得票数 4
我已经检查了两个类似的问题，但它没有帮助。
node 1 [/appl/node1/confluent-4.0.0] ./bin/confluent status elasticsearch-sink
{"name":"elasticsearch-sink","connector":{"state":"RUNNING","worker_id":"10.192.226.24:8083"},"tasks":[{"state":"FAILED&#
浏览 2提问于2018-04-17得票数 0
我有模式注册表容器http://registry-server:8081
ProducerConfig
bootstrap.servers : [PLAINTEXT://kafka-server:9092]
value.serializer : class org.apache.kafka.common.serialization.ByteArraySerializer
以及充当生产者的Standalone service，其属性设置如下
生产者
"schema.registry.url", "http://registry-server:80
浏览 7提问于2022-09-02得票数 1
回答已采纳
我想知道我是否可以使用合流模式注册表生成(然后将其发送到kafka)模式，而不是avro记录？如果是的话，谁能分享一些资源吗？我无法在汇合网站和谷歌上找到任何例子。
我有一个普通分隔的文件，并且有一个单独的模式，目前我正在使用Avro通用记录模式来序列化Avro记录并通过Kafka发送它。这样，模式仍然与记录相关联，这使得它更加庞大。我的逻辑是，如果我删除模式，同时从kafka发送记录，我将能够获得更高的吞吐量。
浏览 1提问于2017-07-13得票数 0
回答已采纳
1回答
我正在尝试设置一个本地kafka -连接堆栈与坞-撰写，我有一个问题，我的scala生产者，应该发送avro消息到一个卡夫卡主题使用模式注册表。
在我的生产者(scala)代码中，我执行以下操作：
val kafkaBootstrapServer = "kafka:9092"
val schemaRegistryUrl = "http://schema-registry:8081"
val topicName = "test"
val props = new Properties()
props.put("boots
浏览 4提问于2021-01-15得票数 0
回答已采纳
在生成/使用者消息时，添加模式注册表的附加层(也称故障点)有什么好处吗？如果服务曾经中断，那么消息就不会被消耗或产生。如果使用Kafka的系统不使用少一个故障点的，岂不是更容易出错吗？
浏览 5提问于2016-10-03得票数 2
回答已采纳
嘿，我正在做KSQL快速入门的例子。问题是我想使用avro格式生成数据，但它抛出了我在底部列出的错误。
该教程位于
复制问题
git clone https://github.com/confluentinc/ksql.git
cd ksql
git checkout 5.5.0-post
cd docs/tutorials/
docker-compose up -d
如果您停靠ps，您应该看到以下容器正在运行：
confluentinc/ksqldb-examples:5.5.0
confluentinc/cp-ksql-server:5.4.0
confluentinc/cp-sch
浏览 0提问于2020-06-06得票数 0
我已经使用Helm和Minikube部署了Kafka。我需要建立一个生产者在Scala的代理IP地址和主机是必需的。我已经定义了NodePort服务来向外部世界公开Kafka。我将代理设置为minkube-ip:service-node-port，但是，我得到了连接异常。我定义的配置有什么问题？有了docker-compose文件，应用程序就能正常工作。 错误堆栈跟踪： Exception in thread "main" org.apache.kafka.common.errors.SerializationException: Error serializing Avro
浏览 54提问于2019-01-22得票数 0
回答已采纳
2回答
我创建了下面的测试类，以便使用AvroSerializer生成一个事件。
@SpringBootTest
@EmbeddedKafka(partitions = 1, brokerProperties = { "listeners=PLAINTEXT://localhost:9092", "port=9092" })
@TestPropertySource(locations = ("classpath:application-test.properties"))
@ContextConfiguration(classes = { TestApp
浏览 6提问于2021-03-11得票数 4
1回答
我正在探索将ITSM系统与Kafka集成的各种选择。这需要Protobuf消息通过rest代理API发送到Kafka。
我遵循了以下两篇文章，并设置了请求主体和标题。
在Confluent上，我创建了一个主题和一个测试模式。虽然我能够通过JSON生成主题下的消息，但我无法生成Protobuf消息。
我甚至无法测试‘GET’方法，因为我无法找到SchemaRegistryHost和端口。如何从汇合控制台获得这些详细信息？
虽然我手动获取架构ID并在输入中提供它，但我收到了消息“{”error_code“：405，”message“：”HTTP405方法不允许“}”。这让我想到，在Schema
浏览 4提问于2022-09-18得票数 0
回答已采纳
我设置了一个Kafka接收器来将事件发送到PostgreSQL。我编写了这个简单的生成器，它向主题发送带有模式(avro)数据的JSON，如下所示：
producer.py (kafka-python)
biometrics = {
"heartbeat": self.pulse, # integer
"oxygen": self.oxygen,# integer
"temprature": self.temprature, # float
"time": time
浏览 2提问于2021-02-14得票数 0
回答已采纳
2回答
我已经在Ubuntu16.04机器上安装了汇合平台，最初我已经配置了动物园管理员、Kafka和ksql，并启动了合流平台。我能看到下面的信息。
root@DESKTOP-DIB3097:/opt/kafkafull/confluent-5.1.0/bin# ./confluent start
This CLI is intended for development only, not for production
https://docs.confluent.io/current/cli/index.html
Using CONFLUENT_CURRENT: /tmp/confluen
浏览 2提问于2019-01-30得票数 5
1回答
消息不以神奇字节开头。
、、、、
我正在尝试使用Go中的/linkedin/goavro包将avro编码的数据生成到kafka主题中。目标是能够使用不同的客户端来使用这个主题。
首先，我将模式注册如下：
curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" --data '{"schema": "{\"name\":\"test_topic2\",\"type\":\"record\", \"fields\":[{
浏览 13提问于2020-03-09得票数 1
我是NiFi和卡夫卡的新手，我已经为这个问题挣扎了几天。我有一个NiFi数据流，以JSON记录被发布到一个Kafka主题结束，它使用配置了一个JSONRecordSetWriter服务的PublishKafkaRecord_2_0处理器作为编写器。一切看起来都很好:消息被发布到Kafka，在发布后查看流文件中的记录，看起来就像格式良好的JSON。不过，在使用命令行上的消息时，我发现它们都是由一个字母加在前面的。尝试使用配置为ConsumeKafkaRecord_2_0的JSONTreeReader读取消息，当然，请参阅错误。
当我尝试了不同的事情时，字母发生了变化:它从"h“开始，然后
浏览 3提问于2018-12-05得票数 1
回答已采纳
当我试图把卡夫卡的话题放进Postgres的桌子上时，这让我抓狂了。这是我的设置，我不知道我做错了什么。
这是卡夫卡主题的典型信息。
{
"flightId": "5cbc7ad25732ab0004c51c45",
"recordedAt": "2022-03-26T18:17:11.356Z",
"device": "iOS",
"platform": "A5",
"vehicleId": "621c12a9b12161
浏览 17提问于2022-03-26得票数 1
回答已采纳
1回答
我试图在Ubuntu上运行一个星火流应用程序，但是我遇到了一些错误。由于某种原因，Ubuntu22.04没有找到jar文件，尽管Windows上的配置是相同的。
我在脚本中运行以下配置
spark = SparkSession \
.builder \
.appName("File Streaming PostgreSQL") \
.master("local[3]") \
.config("spark.streaming.stopGracefullyOnShutdown",
浏览 6提问于2022-08-15得票数 0
回答已采纳
1回答
Avro指纹模式
、、
我正在使用C#中的Confluent kafka来使用消息，这些消息被格式化为十六进制字符串，从中提取模式指纹。如何从C#中的模式指纹中获取模式？我是不是遗漏了什么？
浏览 1提问于2021-07-07得票数 2
我在远程服务器上使用Kafka Debezium MySQL连接器。我一直收到两条错误消息，它们随机出现(我没有编辑任何内容)。
第一个是
[2021-02-26 14:08:00,523] INFO Kafka commitId: 9c1fbb3db1e0d69d (org.apache.kafka.common.utils.AppInfoParser:118)
[2021-02-26 14:08:00,523] INFO Kafka startTimeMs: 1614370080522 (org.apache.kafka.common.utils.AppInfoParser:119)
[2
浏览 6提问于2021-02-27得票数 1
1回答
第一次使用Debezium时，我在debezium中每分钟只收到大约1000条消息(与在线基准相比非常慢)。没有节流卡夫卡连接/ MySQL/ Kafka经纪人，不知道我在这里做什么。我将在这里张贴配置，以供参考。
Kafka的配置-连接工作人员：
-e CONNECT_GROUP_ID="quickstart" \
-e CONNECT_CONFIG_STORAGE_TOPIC="quickstart-config" \
-e CONNECT_OFFSET_STORAGE_TOPIC="quickstart-offsets" \
-e CONN
浏览 7提问于2021-06-14得票数 1
我必须将记录从Aurora/Mysql发送到MSK，然后从那里发送到弹性搜索服务。
Aurora ->kafka- connect ->AWS MSK-->kafka连接-弹性搜索
Aurora表结构中的记录如下所示
我认为记录将以这种格式发送到AWS MSK。
"o36347-5d17-136a-9749-Oe46464",0,"NEW_CASE","WRLDCHK","o36347-5d17-136a-9749-Oe46464","<?xml version=""1.0"
浏览 12提问于2020-01-01得票数 1
回答已采纳
1回答
如果我们在Kafka中使用模式注册表，是否每个生产者都需要在每次将记录发送到代理时发送当前版本的Kafka？
如果是，那么额外的开销有什么意义，因为我们已经在每个avro文件中发送模式了？
如果不是，请为我的问题的愚蠢感到遗憾，并请帮助我更好地理解模式注册表。
浏览 0提问于2017-10-31得票数 0
(end goal)在测试我是否最终可以使用火花流从汇合平台读取avro数据之前，如下所述：
我将验证是否可以使用下面的命令来读取它们：
$ kafka-avro-console-consumer \
> --topic my-topic-produced-using-file-pulse-xml \
> --from-beginning \
> --bootstrap-server localhost:9092 \
> --property schema.registry.url=http://localhost:8081
我收到这个错误消息，未知魔法字节
Proce
浏览 0提问于2020-09-10得票数 0
回答已采纳
我正在使用Kafka_2.12-3.0.0在Ubuntu上以独立模式将主题流到PosgreSQL，并获得反序列化错误。
使用package中的confluent_kafka生成python中的kafka流(works )：
{"pmu_id": 2, "time": 1644329854.08, "stream_id": 2, "stat": "ok", "ph_i1_r": 27.682000117654074, "ph_i1_j": -1.546410917622178,
浏览 12提问于2022-02-08得票数 0
回答已采纳
1回答
我有一个应用程序，它周期性地用原始JSON生成一系列消息。我可以用Avro工具把它转换成avro。我这样做是因为由于Kafka接收器的限制，我需要消息包含模式。我可以在notepad++上打开这个文件，并看到它包括模式和几行数据。
现在，我想将它发送给我的中心Kafka代理，然后使用Kafka接收器将数据放入数据库。我很难理解我应该如何将这些阿夫罗文件发送给我的卡夫卡经纪人。我是否需要一个模式注册表来实现我的目的？我相信Kafkacat不支持Avro，所以我想我必须坚持卡夫卡-生产者. so的卡夫卡安装(请纠正我，如果我错了)。
问题是:能否有人分享步骤，以生产我的Avro文件给一个卡夫卡经纪
浏览 3提问于2021-03-31得票数 0
回答已采纳
我正在使用带有MockSchemaRegistryClient的为我们的kafka系统构建一个集成测试。我正在为我们的一个Stream拓扑构建一个测试，该拓扑是使用Streams API (KStreamBuilder)构建的。此特定拓扑将KStream (stream1)送入KTable (table1)。
我在将来自表1的KTableProcessor的输入馈送到stream1时遇到错误：
Exception in thread "mortgage-kafka-consumers-it-c1dd9185-ce16-415c-ad82-293c1281c897-StreamThrea
浏览 2提问于2018-06-19得票数 5
我正在使用合流模式注册中心的社区版本。据我所知，太多相同主题下的模式可能会在以后引起兼容性问题。我尝试使用以下链接中提到的方法来删除模式注册表中无用的模式：
之后，主题(主题)为空(curl -X获取模式-注册表:8081/ subject /，返回[])，但模式仍然存在(curl -X GET模式-注册表:8081/-X/ids/25，返回模式)。我甚至试着删除kafka-topics.sh中的主题。但是，模式似乎仍然存在，并且可以通过id检索。
有谁遇到过同样的问题吗？我很好奇是否有一种方法可以删除架构。感谢你的帮助！
浏览 0提问于2019-08-21得票数 3
2回答
我在Kubernetes集群的容器中运行了confluent kafka、zookeeper、schema-registry和ksql。Kafka，zookeeper和模式注册表工作得很好，可以创建主题并以Avro格式写入数据，但当我尝试使用curl检查ksql并创建一些流时，例如：
curl -XPOST http://ksql-svc.someapp:8080/ksql -H "Content-Type: application/json" -d $'
{"ksql": "CREATE STREAM kawabanga_stream (lo
浏览 0提问于2018-03-18得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
云直播活动推荐
腾讯云开发者
