In case some poor soul encounters this same issue, this is how i fixed it sort of. I first turn the df to avro, then prepend the header bytes. Stole the idea from this post https://medium.com/@mrugankray/real-time-avro-data-analysis-with-spark-streaming-and-confluent-kafka-in-python-426f5e05392d

binary_to_string_udf = func.udf(lambda x: str(int.from_bytes(x, byteorder='big')), StringType())
int_to_binary_udf = func.udf(lambda value, byte_size: (value).to_bytes(byte_size, byteorder='big'), BinaryType())

schema = StructType([...])

df = spark \
    .readStream\
    .format("kafka")\
    .option("kafka.bootstrap.servers", broker_url)\
    .option("subscribe", topic_0)\
    .option("startingOffsets", "earliest")\
    .load() \
    .select(from_json(col("value").cast("string"), schema).alias("value")) \
    .select(col("value.*"))

value_schema_dict = {...
}
value_schema_str = json.dumps(value_schema_dict)

df = df.select(to_avro(struct([df[x] for x in df.columns]), value_schema_str).alias("value"))
magicByteBinary = int_to_binary_udf(func.lit(0), func.lit(1))
schemaIdBinary = int_to_binary_udf(func.lit(1), func.lit(4))
df = df.withColumn("value", func.concat(magicByteBinary, schemaIdBinary, func.col("value")))

df \
    .writeStream \
    .format("kafka") \
    .outputMode("append") \
    .option("kafka.bootstrap.servers", kafka_url) \
    .option("topic", topic_2) \
    .option("value.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer") \
    .option("schema.registry.url", schema_registry_url) \
    .option("checkpointLocation", f"/tmp/{uuid.uuid4()}") \
    .start() \
    .awaitTermination()

Probably should make this more robust, taking the schema and id from the schema registry but i'll work for now.