如何使用pyspark文件系统读取最近一小时内上传的新文件？

j9per5c4 于 12个月前发布在 Spark

关注(0)|答案(1)|浏览(118)

我试图读取最新的文件（说在过去一个小时的新文件）在一个目录中可用并加载该数据.我正在尝试与pyspark结构化流.我已经尝试了Spark流的maxFileAge选项，但它仍然是加载目录中的所有文件，无论在选项中指定的时间.

spark.readStream\
.option("maxFileAge", "1h")\
.schema(cust_schema)\
    .csv(upload_path) \
    .withColumn("closing_date", get_date_udf_func(input_file_name()))\
    .writeStream.format('parquet') \
    .trigger(once=True) \
    .option('checkpointLocation', checkpoint_path) \
    .option('path', write_path) \
    .start()

字符串
上面是我尝试的代码，但它会加载所有可用的文件，而不管时间。请指出我在这里做错了什么。

pyspark

来源：https://stackoverflow.com/questions/71405715/how-to-read-new-files-uploaded-in-last-one-hour-using-pyspark-filestream

1条答案

按热度按时间

plicqrtu1#

抱歉，来晚了。
.option("maxFileAge", "1h")仅在创建checkpointLocation之后才有效。
范例：
第一次运行，读取所有内容。第二次运行，检查checkpointLocation是否已创建，然后通过maxFileAge过滤并将这些文件添加到checkpointLocation。
你可以用basepath在readstream中只读取一个文件来创建一个checkpointLocation，然后maxage参数将在接下来的批处理中起作用。

spark.readStream\
.option("maxFileAge", "1h")\
.option("basePath", upload_path)\
.schema(cust_schema)\
    .csv(upload_path + "/partition1=xxxx/partition2=yyyy/") \
...

字符串

赞(0）回复(0）举报 12个月前

我来回答

如何使用pyspark文件系统读取最近一小时内上传的新文件？

1条答案

相关问题

热门标签

最新问答