Spark structured streaming read from kafka json编码问题

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder()
  .appName("Streaming from Kafka")
  .master("local[*]")
  .getOrCreate()

import org.apache.spark.sql.functions._
import org.apache.spark.sql.types._

import spark.implicits._

val schema = new StructType()
  .add("field1", StringType)
  .add("field2", IntegerType)
  .add("field3", DoubleType)

val kafkaDF = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "kafka_servers")
  .option("subscribe", "topic_name")
  .option("startingOffsets", "earliest")
  .load()
  .select(from_json(col("value").cast("string"), schema) as "data")
  .select("data.*")

val query = kafkaDF.writeStream
  .format("console")
  .start()

query.awaitTermination()