问为什么流查询不向HDFS写入数据？

EN

val sparkSession = SparkSession
.builder
.appName("xxx")
.config("spark.serializer", 
  "org.apache.spark.serializer.KryoSerializer")
.config("spark.rpc.netty.dispatcher.numThreads", "2")
.config("spark.shuffle.compress", "true")
.config("spark.rdd.compress", "true")
.config("spark.sql.inMemoryColumnarStorage.compressed", "true")
.config("spark.io.compression.codec", "snappy")
.config("spark.broadcast.compress", "true")
.config("spark.sql.hive.thriftServer.singleSession", "true")
.config("hive.exec.dynamic.partition", "true")
.config("hive.exec.dynamic.partition.mode", "nonstrict")
.config("spark.streaming.receiver.writeAheadLog.enable","true")
.enableHiveSupport()
.getOrCreate()

val rawStreamDF = sparkSession
.readStream
.format("kafka")
.option("kafka.bootstrap.servers", <value>)
.option("subscribe", <value>)
.option("key.serializer", <value>)
.option("value.serializer", <value>)
.option("startingOffsets", "earliest")
.option("auto.offset.reset",earliest)
.option("group.id",  <value>)
.option("fetchOffset.numRetries", 3)
.option("fetchOffset.retryIntervalMs", 10)
.option("IncludeTimestamp", true)
.option("enable.auto.commit",  <value>)
.option("security.protocol",  <value>)
.option("ssl.keystore.location",  <value>)
.option("ssl.keystore.password",  <value>)
.option("ssl.truststore.location",  <value>)
.option("ssl.truststore.password",  <value>)
.load()
.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
.as[(String, String)]

val query = rawStreamDF
  .writeStream
  .format("json")
  .option("startingOffsets", "latest")
  .option("path", "STREAM_DATA_PATH")
  .option("checkpointLocation", "checkpointPath")
  .trigger(Trigger.ProcessingTime("5 seconds"))
  .outputMode("append")
  .start

Logger.log.info("Status:"+query.status)
print("Streaming Status1:"+query.status)

query.awaitTermination(450)

Status:{ "message" : "Initializing sources", "isDataAvailable" : false, "isTriggerActive" : false }