Avro and Pydantic
KafkaEgress writes every record as JSON by default. That suits lifecycle events and telemetry, but a consumer that needs a compact, schema-checked format, such as a machine learning service reading predictions, can read Avro instead.
Each topic can have its own serializer, so JSON and Avro topics share one producer, and a Pydantic model can be published as it is.
The code below publishes to Kafka, so start a broker first with odctl up kafka-lite. See Installation for the one-time odctl install. Stop it with odctl down kafka-lite --volumes.
Publish a Pydantic Model
publish_event turns a Pydantic V2 model into a JSON-ready dict before any egress sees it, so you do not convert it yourself. Pydantic V1 models, including pydantic.v1, are not supported and raise a PydanticSerializationError.
from pydantic import BaseModel
from desflow import DynamicRealtimeEnvironment
class TaskEvent(BaseModel):
path_id: str
status: str
def work_task(env: DynamicRealtimeEnvironment, task_id: int):
# The model is passed as it is; publish_event turns it into a dict.
event = TaskEvent(path_id="Shop.packers", status="queued")
env.publish_event(f"task-{task_id}", event)
yield env.timeout(0)
Avro for Some Topics
A topic router picks the topic of each record, and topic_serializers names the serializer of each topic. A topic with no serializer in the mapping uses the default, JsonSerializer.
KafkaEgress serializes the whole record, not only its value: key, sim_ts, timestamp and value, without stream_type. An Avro schema therefore describes that record, with your payload as the value field. ConfluentAvroSerializer uses a Confluent Schema Registry, and GlueAvroSerializer uses an AWS Glue Schema Registry.
from desflow import KafkaEgress
from desflow.connectors.egress.kafka import ConfluentAvroSerializer
# An Avro serializer for the predictions topic. The schema describes the whole record
# KafkaEgress sends, with the prediction as its value.
prediction_serializer = ConfluentAvroSerializer(
registry_url="http://127.0.0.1:8081",
schema_str="""{
"type": "record", "name": "PredictionEvent", "namespace": "com.desflow.ml",
"fields": [
{"name": "key", "type": "string"},
{"name": "sim_ts", "type": "double"},
{"name": "timestamp", "type": "string"},
{"name": "value", "type": {
"type": "record", "name": "MLPrediction",
"fields": [
{"name": "event_type", "type": "string"},
{"name": "task_id", "type": "string"},
{"name": "confidence", "type": "double"}
]
}}
]
}""",
)
# A topic router that sends predictions to their own topic.
def ml_topic_router(data: dict) -> str:
if data.get("stream_type") == "telemetry":
return "sim-telemetry"
value = data.get("value", {})
if value.get("event_type") == "prediction":
return "ml-predictions"
return "sim-events"
egress = KafkaEgress(
bootstrap_servers="localhost:9092",
topic_router=ml_topic_router,
# Only ml-predictions is Avro; sim-telemetry and sim-events use JSON.
topic_serializers={"ml-predictions": prediction_serializer},
)
Generate the Schema from the Model
A schema written by hand as a JSON string can drift from the model it describes. The dataclasses-avroschema package generates it from a Pydantic model instead. AvroBaseModel is a Pydantic model that can also write its Avro schema:
from dataclasses_avroschema.pydantic import AvroBaseModel
from pydantic import Field
class MLPrediction(AvroBaseModel):
"""The prediction, published as the event value."""
event_type: str = "prediction"
task_id: str = Field(...)
confidence: float = Field(...)
class PredictionRecord(AvroBaseModel):
"""The record KafkaEgress serializes for an event."""
key: str
sim_ts: float
timestamp: str
value: MLPrediction
class Meta:
namespace = "com.desflow.ml"
schema_name = "PredictionEvent"
schema_string = PredictionRecord.avro_schema()
Both ConfluentAvroSerializer and GlueAvroSerializer take the generated string as schema_str:
generated_serializer = ConfluentAvroSerializer(
registry_url="http://127.0.0.1:8081",
schema_str=PredictionRecord.avro_schema(), # follows the model
)
egress = KafkaEgress(
bootstrap_servers="localhost:9092",
topic_router=ml_topic_router,
topic_serializers={"ml-predictions": generated_serializer},
)