I'm a newbie to Avro. I'm trying to parse a simple CSV file containing one string value and one int value, but I'm getting the error: avro.io.AvroTypeException: The datum is not an example of the schema
The schema I'm using is:
{"namespace": "paymenttransaction",
"type": "record",
"name": "Payment",
"fields": [
{"name": "TransactionId", "type": "string"},
{"name": "Id", "type": "int"}
]
}
The CSV file has the content:
TransactionId,Id
2018040101000222749,1
And the Python code that I am running for the producer is:
from confluent_kafka import avro
from confluent_kafka.avro import AvroProducer
import csv
value_schema = avro.load('/home/daniela/avro/example.avsc')
AvroProducerConf = {'bootstrap.servers': 'localhost:9092',
'schema.registry.url': 'http://localhost:8081',
}
avroProducer = AvroProducer(AvroProducerConf, default_value_schema=value_schema)
with open('/home/usertest/avro/data/paymenttransactions.csv') as file:
reader = csv.DictReader(file, delimiter=",")
for row in reader:
avroProducer.produce(topic='test', value=row)
print(row)
avroProducer.flush()
What am I doing wrong?
It's because Id is still a string and schema requires an int.
Try with :
with open('/home/usertest/avro/data/paymenttransactions.csv') as file:
reader = csv.DictReader(file, delimiter=",")
for row in reader:
data_set = {"TransactionId": row["TransactionId"], "Id": int(row["Id"])}
avroProducer.produce(topic='test', value=data_set)
print(row)
avroProducer.flush()