Spark甚至在分区数据中列出所有叶节点

Question

我有date&hour,文件夹结构分区的镶木地板数据:

events_v3
  -- event_date=2015-01-01
    -- event_hour=2015-01-1
      -- part10000.parquet.gz
  -- event_date=2015-01-02
    -- event_hour=5
      -- part10000.parquet.gz

我已经raw_events通过spark 创建了一个表,但是当我尝试查询时,它会扫描所有目录的页脚,这会减慢初始查询,即使我只查询一天的数据.

查询: select * from raw_events where event_date='2016-01-01'

类似的问题:http://mail-archives.apache.org/mod_mbox/spark-user/201508.mbox/%3CCAAswR-7Qbd2tdLSsO76zyw9tvs-Njw2YVd36bRfCG3DKZrH0tw@mail.gmail.com%3E(但它的旧版本)

日志:

App > 16/09/15 03:14:03 main INFO HadoopFsRelation: Listing leaf files and directories in parallel under: s3a://bucket/events_v3/

然后它产生350个任务,因为有350天的数据.

我已经禁用了schemaMerge,并且还指定了架构来读取,因此它可以转到我正在查看的分区,为什么要打印所有叶子文件？列出具有2个执行程序的叶子文件需要10分钟,查询实际执行需要20秒

代码示例:

val sparkSession = org.apache.spark.sql.SparkSession.builder.getOrCreate()
val df = sparkSession.read.option("mergeSchema","false").format("parquet").load("s3a://bucket/events_v3")
    df.createOrReplaceTempView("temp_events")
    sparkSession.sql(
      """
        |select verb,count(*) from temp_events where event_date = "2016-01-01" group by verb
      """.stripMargin).show()

Answer 1

一旦给出了一个要从中读取的目录,就会调用listLeafFiles(org/apache/spark/sql/execution/datasources/fileSourceInterfaces.scala).这反过来调用fs.listStatusapi调用来获取文件和目录列表.现在,对于每个目录,再次调用此方法.这是递归的,直到没有目录.这种设计在HDFS系统中运行良好.但是在s3中工作不好,因为列表文件是RPC调用.其他S3支持通过前缀获取所有文件,这正是我们所需要的.

因此,例如,如果我们的目录结构具有1年的数据,每个目录为小时和10个子目录,我们将有365*24*10 = 87k api调用,这可以减少到138 api调用,因为有只有137000个文件.每个s3 api调用返回1000个文件.

码: org/apache/hadoop/fs/s3a/S3AFileSystem.java

public FileStatus[] listStatusRecursively(Path f) throws FileNotFoundException,
            IOException {
        String key = pathToKey(f);
        if (LOG.isDebugEnabled()) {
            LOG.debug("List status for path: " + f);
        }

        final List<FileStatus> result = new ArrayList<FileStatus>();
        final FileStatus fileStatus =  getFileStatus(f);

        if (fileStatus.isDirectory()) {
            if (!key.isEmpty()) {
                key = key + "/";
            }

            ListObjectsRequest request = new ListObjectsRequest();
            request.setBucketName(bucket);
            request.setPrefix(key);
            request.setMaxKeys(maxKeys);

            if (LOG.isDebugEnabled()) {
                LOG.debug("listStatus: doing listObjects for directory " + key);
            }

            ObjectListing objects = s3.listObjects(request);
            statistics.incrementReadOps(1);

            while (true) {
                for (S3ObjectSummary summary : objects.getObjectSummaries()) {
                    Path keyPath = keyToPath(summary.getKey()).makeQualified(uri, workingDir);
                    // Skip over keys that are ourselves and old S3N _$folder$ files
                    if (keyPath.equals(f) || summary.getKey().endsWith(S3N_FOLDER_SUFFIX)) {
                        if (LOG.isDebugEnabled()) {
                            LOG.debug("Ignoring: " + keyPath);
                        }
                        continue;
                    }

                    if (objectRepresentsDirectory(summary.getKey(), summary.getSize())) {
                        result.add(new S3AFileStatus(true, true, keyPath));
                        if (LOG.isDebugEnabled()) {
                            LOG.debug("Adding: fd: " + keyPath);
                        }
                    } else {
                        result.add(new S3AFileStatus(summary.getSize(),
                                dateToLong(summary.getLastModified()), keyPath,
                                getDefaultBlockSize(f.makeQualified(uri, workingDir))));
                        if (LOG.isDebugEnabled()) {
                            LOG.debug("Adding: fi: " + keyPath);
                        }
                    }
                }

                for (String prefix : objects.getCommonPrefixes()) {
                    Path keyPath = keyToPath(prefix).makeQualified(uri, workingDir);
                    if (keyPath.equals(f)) {
                        continue;
                    }
                    result.add(new S3AFileStatus(true, false, keyPath));
                    if (LOG.isDebugEnabled()) {
                        LOG.debug("Adding: rd: " + keyPath);
                    }
                }

                if (objects.isTruncated()) {
                    if (LOG.isDebugEnabled()) {
                        LOG.debug("listStatus: list truncated - getting next batch");
                    }

                    objects = s3.listNextBatchOfObjects(objects);
                    statistics.incrementReadOps(1);
                } else {
                    break;
                }
            }
        } else {
            if (LOG.isDebugEnabled()) {
                LOG.debug("Adding: rd (not a dir): " + f);
            }
            result.add(fileStatus);
        }

        return result.toArray(new FileStatus[result.size()]);
    }

/org/apache/spark/sql/execution/datasources/fileSourceInterfaces.scala

def listLeafFiles(fs: FileSystem, status: FileStatus, filter: PathFilter): Array[FileStatus] = {
    logTrace(s"Listing ${status.getPath}")
    val name = status.getPath.getName.toLowerCase
    if (shouldFilterOut(name)) {
      Array.empty[FileStatus]
    }
    else {
      val statuses = {
        val stats = if(fs.isInstanceOf[S3AFileSystem]){
          logWarning("Using Monkey patched version of list status")
          println("Using Monkey patched version of list status")
          val a = fs.asInstanceOf[S3AFileSystem].listStatusRecursively(status.getPath)
          a
//          Array.empty[FileStatus]
        }
        else{
          val (dirs, files) = fs.listStatus(status.getPath).partition(_.isDirectory)
          files ++ dirs.flatMap(dir => listLeafFiles(fs, dir, filter))

        }
        if (filter != null) stats.filter(f => filter.accept(f.getPath)) else stats
      }
      // statuses do not have any dirs.
      statuses.filterNot(status => shouldFilterOut(status.getPath.getName)).map {
        case f: LocatedFileStatus => f

        // NOTE:
        //
        // - Although S3/S3A/S3N file system can be quite slow for remote file metadata
        //   operations, calling `getFileBlockLocations` does no harm here since these file system
        //   implementations don't actually issue RPC for this method.
        //
        // - Here we are calling `getFileBlockLocations` in a sequential manner, but it should not
        //   be a big deal since we always use to `listLeafFilesInParallel` when the number of
        //   paths exceeds threshold.
        case f => createLocatedFileStatus(f, fs.getFileBlockLocations(f, 0, f.getLen))
      }
    }
  }

Answer 2

为了澄清 Gaurav 的回答，截断的代码来自 Hadoop 分支 2，可能直到 Hadoop 2.9 才会出现（请参阅HADOOP-13208）；并且有人需要更新 Spark 以使用该功能（这不会损害使用 HDFS 的代码，只是不会在那里显示任何加速）。

需要考虑的一件事是：什么是对象存储的良好文件布局。

• 不要有很深的目录树，每个目录只有几个文件
• 确实有包含许多文件的浅树
• 考虑将文件的前几个字符用于变化最大的值（例如天/小时），而不是最后一个。为什么？一些对象存储似乎使用前导字符进行散列，而不是尾随字符......如果你给你的名字更多的唯一性，那么它们就会分布在更多的服务器上，具有更好的带宽/更少的节流风险。
• 如果您使用的是 Hadoop 2.7 库，请切换到 s3a:// 而不是 s3n://。它已经更快了，并且每周都在变得更好，至少在 ASF 源代码树中是这样。

最后，Apache Hadoop、Apache Spark 和相关项目都是开源的。欢迎投稿。这不仅仅是代码，它是文档、测试，以及针对这些性能的东西，针对您的实际数据集进行测试。甚至向我们提供有关导致问题的原因（以及您的数据集布局）的详细信息也很有趣。