如何在pyspark的结构化流作业中运行地图转换

问题描述

我正在尝试通过进行REST API调用的map（）转换来设置结构化流作业。详细信息如下：

(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（）中执行带有流源的查询。

任何帮助将不胜感激。谢谢。

解决方法

您还必须在df上执行流 意思是df.writeStream.start（）..

这里有一个类似的线程：

Queries with streaming sources must be executed with writeStream.start();