优化从s3存储桶中的分区实木复合地板文件读取

问题描述

我有一个实木复合地板格式的大型数据集（大小约为1TB），分为两个层次结构：CLASS和DATE 只有7个班级。但是日期从2020-01-01起一直在增加。 我的数据先按CLASS分区，然后按DATE

分区

类似这样：

CLASS1---DATE 1
      ---DATE 2
      ---  .
      ---  .
      ---  .
      ---DATE N

CLASS2---DATE 1
      ---DATE 2
      ---  .
      ---  .
      ---  .
      ---DATE N

我通过CLASS在for循环中加载数据。如果加载整个实木复合地板文件，YARN将杀死该作业，因为它会使内存实例超载。但是自从我在建模中执行百分位计算以来，我一直都在加载。此方法大约需要23个小时才能完成。

但是，如果我重新分区以仅拥有CLASS分区，则该作业大约需要10个小时。 子分区过多是否会降低Spark executor的工作速度？ 我之所以将分区层次结构保持为CLASS-> DATE，仅是因为我每天都需要通过DATE附加新数据。 如果只有1个分区更有效，那么在加载新数据后，我每天都必须重新分区为CLASS分区。 有人可以解释为什么拥有一个分区能更快地工作吗？如果是这样，那么每天通过追加而不重新分割整个数据集的最佳方法是什么？

谢谢

编辑： 我在文件结构上使用for循环像这样通过CLASS分区循环：

fs = s3fs.S3FileSystem(anon=False)    
inpath="s3://bucket/file.parquet/"

Dirs= fs.ls(inpath)
for paths in Dirs:
    customPath='s3://' + uvapath + '/'
    class=uvapath.split('=')[1]
    df=spark.read.parquet(customPath)
    outpath="s3://bucket/Output_" + class + ".parquet"
#Perform calculations
df.write.mode('overwrite').parquet(outpath)

已加载的df将具有CLASS=1的所有日期。然后，我为每个CLASS将文件输出为单独的实木复合地板文件，这样我就有7个实木复合地板文件：

Output_1.parquet
Output_2.parquet
Output_3.parquet
Output_4.parquet
Output_5.parquet
Output_6.parquet
Output_7.parquet

然后我将7个实木复合地板合并为一个实木复合地板不是问题，因为生成的实木复合地板文件要小得多。

解决方法

我有包含三列的分区数据，年份，月份和ID。文件夹路径层次结构是

year=2020/month=08/id=1/*.parquet
year=2020/month=08/id=2/*.parquet
year=2020/month=08/id=3/*.parquet
...
year=2020/month=09/id=1/*.parquet
year=2020/month=09/id=2/*.parquet
year=2020/month=09/id=3/*.parquet

我可以通过加载根路径来读取DataFrame。

val df = spark.read.parquet("s3://mybucket/")

然后，已分区的列将自动添加到DataFrame中。现在，您可以通过以下方式过滤分区列的数据

val df_filtered = df.filter("year = '2020' and month = '09'")

并使用df_filtered进行操作，则火花将仅使用分区的数据！

---

对于重复处理，可以使用火花的fair scheduler。使用以下代码，将fair.xml文件添加到项目的src / main / resources中

<?xml version="1.0"?>

<allocations>
    <pool name="fair">
        <schedulingMode>FAIR</schedulingMode>
        <weight>10</weight>
        <minShare>0</minShare>
    </pool>
</allocations>

并在创建spark会话后设置spark配置。

spark.sparkContext.setLocalProperty("spark.scheduler.mode","FAIR")
spark.sparkContext.setLocalProperty("spark.scheduler.allocation.file",getClass.getResource("/fair.xml").getPath)
spark.sparkContext.setLocalProperty("spark.scheduler.pool","fair")

然后，您可以并行完成工作。您可能要根据CLASS并行化作业，所以

val classes = (1 to 7).par
val date = '2020-09-25'

classes foreach { case i =>

    val df_filtered = df.filter(s"CLASS == '$i' and DATE = '$date'")
    
    // Do your job

}

代码将在不同的CLASS值下同时工作。