结构化流式传输到rdd不起作用

试图对stractured流做一些操作，并且需要使用.rdd函数，但得到"""AnalysisException: Queries with streaming sources must be executed with writeStream.start();;"""？

val d = df.select(from_json(col("value").cast("string"), schema).cast("string").alias("url")).as[String]

.rdd.toDF()
.writeStream
.format("console") // <-- use ConsoleSink
.option("truncate", false)
.option("numRows", 10)
.trigger(Trigger.ProcessingTime(5 seconds))
.queryName("rate-console")
.start
.awaitTermination()