我正在尝试使用map（）转换设置一个结构化流式处理作业，该转换可以进行restapi调用。详情如下：

(1)
df=spark.readStream.format('delta') \
.option("maxFilesPerTrigger", 1000) \
.load(f'{file_location}') 

(2)
respData=df.select("resource", "payload").rdd.map(lambda row: put_resource(row[0], row[1])).collect()
respDf=spark.createDataFrame(respData, ["resource", "status_code", "reason"])

(3)
respDf.writeStream \
.trigger(once=True) \
.outputMode("append") \
.format("delta") \
.option("path", f'{file_location}/Response') \
.option("checkpointLocation", f'{file_location}/Response/Checkpoints') \
.start()

但是，我得到了一个错误：带有流源的查询必须在步骤（2）上使用writestream.start（）执行。
任何帮助都将不胜感激。谢谢您。