我有大量的目录,并使用spark分区过滤来读取Dataframe。我发现我直接在文件系统上查找分区目录,我至少可以比spark快一个数量级。我错过什么了吗?为什么spark不并行查找目录?
比如说 /mydata/date=2020-09-01
, /mydata/date=2020-09-02
等等,我只想 2020-09-28
至 2020-09-29
.
我可以:
val paths = List("/mydata/date=2020-09-28", "/mydata/date=2020-09-29")
spark.read.format("parquet").option("basePath", "/mydata").load(paths: _*)
或者,我可以:
val dates = List("2020-09-28", "2020-09-29")
spark.read.format("parquet").load("/mydata").filter($"date".isin(dates: _*))
在第一种情况下,我将不得不自己扫描目录,以确保分区文件存在,否则 load
如果缺少任何一个路径,“找不到文件”将失败。
在第二种情况下,spark将为我扫描目录,并使用分区过滤打开最少数量的文件。
但是,如果目录的数量很大,并且它们位于慢媒体(如s3)上,我可以通过使用并行集合查找目录,比spark快得多。比如:
val goodPaths = paths.par.map(checkExist).toList
我宁愿让spark做spark擅长的事情,但是在一个线程上像这样扫描s3是非常慢的。我是否缺少一些配置选项来允许spark使用节点的所有核心?
我在spark 2.x和3.x上都看到了。
暂无答案!
目前还没有任何答案,快来回答吧!