Kafka 提供商的 Avro Json

Question

对于以下代码，我在行中出错

value_serializer=lambda m: io.DatumWriter(avro_schema).write(m).bytes()

说

TypeError: write() missing 1 required positional argument: 'encoder'

avro write 的写语法应该是什么？

import os
import json
import time
import random
from kafka import KafkaProducer
from dotenv import load_dotenv
from avro import schema, io
from avro.datafile import DataFileWriter

load_dotenv()

BOOTSTRAP_SERVERS = os.getenv("KAFKA_BOOTSTRAP_SERVERS").split(",")
SSL_TRUSTSTORE = os.getenv("KAFKA_SSL_TRUSTSTORE")
TOPIC_NAME = os.getenv("KAFKA_TOPIC")
LINGER_DURATION = int(os.getenv("LINGER_DURATION"))
MESSAGE_BATCH = int(os.getenv("MESSAGE_BATCH"))


def random_with_N_digits(n):
    range_start = 10 ** (n - 1)
    range_end = (10 ** n) - 1
    return random.randint(range_start, range_end)


def produce(message, avro_schema):
    print(f'Sending {message}')
    producer.send(TOPIC_NAME, value=message)


producer = KafkaProducer(
    security_protocol="SSL",
    ssl_cafile=SSL_TRUSTSTORE,
    ssl_check_hostname=False,
    bootstrap_servers=BOOTSTRAP_SERVERS,
    linger_ms=LINGER_DURATION,
    value_serializer=lambda m: io.DatumWriter(avro_schema).write(m).bytes()
)

# Define the Avro schema for the message
schema_str = """
{
    "namespace": "example.avro",
    "type": "record",
    "name": "Message",
    "fields": [
        {"name": "id", "type": "int"},
        {"name": "name", "type": "string"}
    ]
}
"""
avro_schema = schema.Parse(schema_str)


def startStream():
    msg_count = 0

    try:
        while (1):
            # Create a message using the Avro schema
            message = {"id": random.randint(0, 9999999), "name": str(random_with_N_digits(12))}

            # Publish the message to a Kafka topic
            produce(message, avro_schema)

            msg_count += 1
            if msg_count % MESSAGE_BATCH == 0:
                time.sleep(1)

    except KeyboardInterrupt:
        print('interrupted!')
        print(f"Published {msg_count} messages")


startStream()