无法使用python将avro数据写入kafka
Avro是一种数据序列化格式，它使用JSON格式定义数据结构，并提供了强大的数据压缩和快速的序列化/反序列化能力。Kafka是一个分布式流处理平台，它可以处理高吞吐量的实时数据流。
在Python中，可以使用第三方库confluent-kafka来将Avro数据写入Kafka。confluent-kafka是一个功能强大且易于使用的Python客户端，它提供了与Kafka的高效交互。
以下是使用confluent-kafka将Avro数据写入Kafka的步骤：
- 安装
confluent-kafka库： - 安装
confluent-kafka库： - 导入必要的模块：
- 导入必要的模块：
- 定义Avro数据的Schema：
- 定义Avro数据的Schema：
- 创建AvroProducer实例：
- 创建AvroProducer实例：
- 注意，上述代码中的
kafka_server和schema_registry需要替换为实际的Kafka服务器和Schema Registry的地址。 - 构造Avro数据并发送到Kafka：
- 构造Avro数据并发送到Kafka：
- 上述代码中的
my_topic需要替换为实际的Kafka主题。
通过以上步骤，就可以使用Python将Avro数据写入Kafka。这种方法适用于需要使用Avro格式进行数据序列化和反序列化的场景，例如在大数据处理、实时流处理和事件驱动架构中。
腾讯云提供了一系列与Kafka相关的产品和服务，例如Tencent Kafka和Tencent Message Queue for Apache Kafka (CMQ for Kafka)，它们可以帮助用户快速构建可靠的消息传递系统和实时数据处理平台。
相关·内容
1回答
生产者:数据{'..'}不是模式的示例
python、apache-kafka、avro、kafka-producer-api、confluent-platform
我无法为特定模式生成数据，也无法理解原因。代码中包含为字典的示例数据是直接使用汇合的“avro-随机生成器”创建的，因此示例数据必须是正确的，因为它是直接从模式派生的。模式注册表和Avro随机生成器都是合流工具，因此它们的工具不可能产生与它们的模式注册表不兼容的示例数据。
这是模式：
{
"type": "record",
"name": "schemaV1",
"namespace": "net.avro.schemaV1",
"doc": "",
浏览 10提问于2019-10-15得票数 0
3回答
对于AvroProducer到Kafka，“键”和“值”的avro schema在哪里？
apache-kafka、kafka-producer-api、kafka-python
从中的AvroProducer示例可以看出，键/值模式是从文件加载的。也就是说，从下面的代码中：
from confluent_kafka import avro
from confluent_kafka.avro import AvroProducer
value_schema = avro.load('ValueSchema.avsc')
key_schema = avro.load('KeySchema.avsc')
value = {"name": "Value"}
key = {"name":
浏览 1提问于2017-04-28得票数 6
1回答
Avro，Kafka消息键插入无关字符
python、apache-kafka、avro、ksqldb
我的主题是这样构建的 producer = AvroProducer(
config.confluent.connection_config(config.confluent.CONNECTION_MODE_PRODUCER),
default_value_schema=get_avro_schema_object(),
default_key_schema=avro.loads('{"type": "string"}'))
producer.produce(topic=
浏览 22提问于2019-12-10得票数 0
1回答
我是阿夫罗的新手。我试图解析一个包含一个字符串值和一个int值的简单CSV文件，但是我得到了错误：avro.io.AvroTypeException:数据不是模式的示例
我使用的模式是：
{"namespace": "paymenttransaction",
"type": "record",
"name": "Payment",
"fields": [
{"name": "TransactionId", "type":
浏览 0提问于2018-05-11得票数 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
回答已采纳
2回答
反序列化Avro消息
python、apache-kafka、avro、apache-kafka-connect、kafka-producer-api
我从部署了卡夫卡。此外，我还向docker-compose.yml Postgres容器添加了如下内容：
postgres:
image: postgres
hostname: kafka-postgres
container_name: kafka-postgres
depends_on:
- ksql-server
- broker
- schema-registry
- connect
ports:
- 5432:5432
创建了一个主题页面视图。
此外，我创建了带有设置的DatagenC
浏览 0提问于2020-02-01得票数 3
回答已采纳
1回答
无法使用python将avro数据写入kafka
python、apache-kafka、avro、confluent-platform、confluent-schema-registry
我使用kafka kafka_2.11-0.11.0.2和confluent版本3.3.0作为模式注册表。 我定义了一个avro模式，如下所示： {
"namespace": "com.myntra.search",
"type": "record",
"name": "SearchDataIngestionObject",
"fields": [
{"name": "timestamp","type":"long"
浏览 34提问于2020-08-18得票数 1
3回答
confluent-python可以生成以avro为值，以字符串为键的数据吗？
python、apache-kafka
我正在使用python3和向卡夫卡发送消息。我需要发送Avro格式的值和字符串作为关键字的数据。但是我发现confluent-python只能在Avro中同时发送，或者两者都在字符串中发送。confluent-python源代码已经被炸飞了：
def produce(self, **kwargs):
"""
Asynchronously sends a message to Kafka by encoding with specified or default Avro schema.
:param str topic: top
浏览 0提问于2019-02-27得票数 4
1回答
Spark Structured kafka avro制作人
apache-spark、kafka-producer-api、spark-structured-streaming
我有一个数据帧，假设：
val someDF = Seq(
(8, "bat"),
(64, "mouse"),
(-27, "horse")
).toDF("number", "word")
我想使用avro序列化和模式注册表将该数据帧发送到kafka主题。我相信我快到了，但是我似乎不能通过Task not serializable错误。我知道kafka有一个接收器，但它不会与模式注册表通信，这是必需的。
object Holder extends Serializable{
def prop()
浏览 0提问于2018-05-24得票数 0
1回答
无法将后续交易推给卡夫卡制作者
python、apache-kafka、distributed-system、confluent-kafka-python、kafka-transactions-api
我试图将事件分批发送给Kafka (比方说，每批10个事件)，以维护事务性保证。但我只能成功地将第一批事件发送给卡夫卡。所有后续批都失败，原因是KafkaError{code=_STATE，val=-172，str=“在状态就绪中无效”}
下面是我在Kafka中使用事务的代码。
DEFAULT_PRODUCER_CONFIG = {
"bootstrap.servers": BOOTSTRAP_SERVERS,
"schema.registry.url": SCHEMA_REGISTRY_URL,
"enable.idempotence": Tr
浏览 12提问于2022-07-02得票数 0
3回答
TypeError:类型为“mappingproxy”的对象不能被JSON序列化
python、apache-kafka、avro、confluent-schema-registry、confluent-platform
我试图使用模式注册的方式，使用AvroProducer of confluent python发布一条avro消息。但是代码无法序列化枚举类型。下面是代码和错误跟踪。任何帮助都是非常感谢的。
from confluent_kafka import avro
from confluent_kafka.avro import AvroProducer
from example_schema.schema_classes import SCHEMA as value_schema
from example_schema.com.acme import *
import json
def funct
浏览 0提问于2019-07-11得票数 2
回答已采纳
1回答
无法在卡夫卡话题中推动Json
python、json、apache-kafka、avro、confluent-platform
我试图在KAFKA主题中插入JSON格式的数据，但没有成功。
我使用了以下AVRO模式：
{"schemaType":"AVRO","schema":"{\"title\":\"json pipeline\",\"name\":\"MyClass\",\"type\":\"record\",\"namespace\":\"com.acme.avro\",\"fields\":[{\"nam
浏览 4提问于2021-11-05得票数 0
1回答
使用pyinstaller(4.5)和librdkafka 1.7.0将python程序转换为exe应用程序时，找不到错误"confluent_kafka.libs“
python、pyinstaller、librdkafka、confluent-kafka-python
在windows 64位os上的python(python 3.8.11)程序中使用的最新版本的librdkafka 1.7.0运行良好，但当我将程序转换为exe文件并运行时，将得到以下错误
File "produce-simple-avro.py", line 1, in
File "", line 991, in _find_and_load
File "", line 975, in find_and_load_unlocked
File "", line 671, in load_unlocked
File "Py
浏览 7提问于2021-08-09得票数 2
2回答
无法从对象生成avro通用记录
java、apache-kafka、avro、confluent-platform
我正在尝试发送avro记录到一个卡夫卡主题使用卡夫卡生产者。我有一个User类，我正在发送该类的对象。如果我使用avroRecord.put();来设置每个属性，下面的代码可以很好地工作。但我想要的是从一个对象创建一个通用记录，而不使用avroRecord.put()；为每个属性。
用户类
public class User {
int id;
String name;
public User(int id, String name) {
super();
this.id = id;
this.name = na
浏览 3提问于2018-04-24得票数 1
1回答
如何使用Python在Kafka模式注册表中编程注册Avro模式
python、apache-kafka、avro、confluent-schema-registry、kafka-python
我使用python将数据和模式放到kafka和模式注册表中。
from confluent_kafka import avro
from confluent_kafka.avro import AvroProducer
value_schema_str = """
{
"type":"record",
"name":"myrecord",
"fields":[
{
"n
浏览 2提问于2020-04-29得票数 3
回答已采纳
2回答
如何用Python制作卡夫卡墓碑阿夫罗唱片？
python、apache-kafka、kafka-producer-api、kafka-python
我的接收器属性:
{
"name": "jdbc-oracle",
"config": {
"connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
"tasks.max": "1",
"topics": "orders",
"connection.url": "jdbc:oracle:thin:@10.1.2.3:
浏览 3提问于2020-04-13得票数 3
回答已采纳
1回答
尝试解析可为空的字段时获取AvroTypeException
java、json、parsing、avro
我从调用/subjects/{subject}/versions/latest的Kafka schema Registry API获取avro模式，其中有一个"type":["null","string"]字段
我用这个模式创建了一个新的JsonDecoder和DatumReader：
JsonDecoder jsonDecoder = DecoderFactory.get().jsonDecoder(schema, body);
DatumReader<Object> reader = new GenericDatumReader&
浏览 24提问于2021-11-30得票数 0
1回答
如何将水磨机中的unmarshaler (lib for Kafka)改为avro
java、go、apache-kafka、avro
我正在尝试调试为什么不可能反序列化我们通过Kafka接收的数据，并基于一些特定的模式使用avro序列化。
它看起来像这样。有一个用java编写的生产者，它根据某种avro模式序列化数据，将数据发送到Kafka主题，而在消费者端有一个用Golang编写的程序。
但每次我尝试反序列化该数据时，我都会收到错误消息，指出无法反序列化，因为字符串的长度小于零。(依赖项字段)
模式如下所示：
{
"namespace": "data.avro",
"doc":"Docstring.",
"name":
浏览 1提问于2021-09-12得票数 1
5回答
用于在没有KafkaAvroSerializer的情况下序列化Avro的
java、apache-kafka、avro、confluent-schema-registry
我是卡夫卡和阿夫罗的菜鸟。所以我一直在努力让生产者/消费者运行。到目前为止，我已经能够使用以下方法生成和使用简单的Bytes和String:生产者的配置：
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
浏览 8提问于2017-08-11得票数 27
回答已采纳
1回答
创建使用模式但没有模式注册表url的Kafka生产者
c#、apache-kafka、confluent-platform、confluent-schema-registry
下午好，我最近才开始和卡夫卡合作，我有一个关于制片人的问题。
最初，我试图在C#中构建一个没有模式的简单生产者。到目前为止，这是工作的，代码也提供了一个简短的版本。
没有模式的生产者代码：
var config = new ProducerConfig
{
BootstrapServers = "localhost:9092",
BrokerAddressFamily = BrokerAddressFamily.V4,
};
using (var producer = new ProducerBuilder<Null, string>(config
浏览 0提问于2020-10-21得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
即时通信 IM
对象存储活动推荐
腾讯云开发者
