df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")

   .write
   .format("kafka")
   .option("kafka.bootstrap.servers",
           "host1:port1,host2:port2")
   .option("topic", "topic1")
   .save()