我想从按年/月/日/小时划分的s3存储桶中提取指定的天数。这个bucket每天都会添加新文件,而且会变得相当大。我想做什么 spark.read.parquet(<path>).filter(<condition>)
但是,当我运行它时,它比指定路径花费的时间(0.5小时)要长得多(1.5小时)。我不明白为什么要花更长的时间,我应该添加一个 .partitionBy()
从桶里读东西的时候?还是因为bucket中有大量的数据需要过滤?
我想从按年/月/日/小时划分的s3存储桶中提取指定的天数。这个bucket每天都会添加新文件,而且会变得相当大。我想做什么 spark.read.parquet(<path>).filter(<condition>)
但是,当我运行它时,它比指定路径花费的时间(0.5小时)要长得多(1.5小时)。我不明白为什么要花更长的时间,我应该添加一个 .partitionBy()
从桶里读东西的时候?还是因为bucket中有大量的数据需要过滤?
1条答案
按热度按时间6mw9ycah1#
您面临的问题是关于分区发现的。如果您指向Parquet文件所在的路径
spark.read.parquet("s3://my_bucket/my_folder")
spark将在任务管理器中触发一个名为这是一种分区发现方法。为什么会这样?当您使用路径进行调用时,spark无法找到分区的位置以及有多少分区。
在我的情况下,如果我这样计算:
它将触发列表,大约1700个文件夹需要19秒。加上7秒,总共有26秒。
要解决这个开销时间,您应该使用元存储。aws用aws胶水提供了一个很好的解决方案,就像hadoop环境中的hivemetastore一样使用。
使用glue可以存储表元数据和所有分区。而不是给出Parquet地板路径,而是指向table,就像这样:
对于相同的数据,使用相同的过滤器。列表文件不存在,整个计数过程只用了9秒。
在你的情况下,你按年、月、日、时划分。我们每年讨论8760个文件夹。
我建议你看看这个链接和这个链接
这将展示如何使用胶水作为您的Hive元存储。这对提高分区查询的速度有很大的帮助。