如何用apache spark处理数百万个较小的s3文件

Question

sc.wholeTextFiles(s3aPath + "/*/*/*/*.txt").count();

注意：计数是对处理文件需要多长时间的更多调试。这项工作几乎花了一整天的时间，超过10个实例，但仍然失败，错误发布在列表的底部。然后我找到了这个链接，它基本上说这不是最佳的:https://forums.databricks.com/questions/480/how-do-i-ingest-a-large-number-of-files-from-s3-my.html

然后，我决定尝试另一个我目前找不到的解决方案，即加载所有路径，然后联合所有rdds

    ObjectListing objectListing = s3Client.listObjects(bucket);
    List<JavaPairRDD<String, String>> rdds = new ArrayList<>();
    List<JavaPairRDD<String, String>> tempMeta = new ArrayList<>();
    //initializes objectListing
    tempMeta.addAll(objectListing.getObjectSummaries().stream()
            .map(func)
            .filter(item -> item != null && item.getMediaType().equalsIgnoreCase("transcript"))
            .map(item -> SparkConfig.getSparkContext().wholeTextFiles("s3a://" + bucket + "/" + item.getFileName()))
            .collect(Collectors.toList()));

    while(objectListing.isTruncated()) {
        objectListing = s3Client.listNextBatchOfObjects(objectListing);
        tempMeta.addAll(objectListing.getObjectSummaries().stream()
                .map(func)
                .filter(item -> item != null && item.getMediaType().equalsIgnoreCase("transcript"))
                .map(item -> SparkConfig.getSparkContext().wholeTextFiles("s3a://" + bucket + "/" + item.getFileName()))
                .collect(Collectors.toList()));
        if (tempMeta.size() > 5000) {
            rdds.addAll(tempMeta);
            tempMeta = new ArrayList<>();
        }
    }

    if (!tempMeta.isEmpty()){
        rdds.addAll(tempMeta);
    }
    return SparkConfig.getSparkContext().union(rdds.get(0), rdds.subList(1, rdds.size()));

然后，即使将emrfs-site配置设置为：

{
    "Classification": "emrfs-site",
    "Properties": {
      "fs.s3.consistent.retryPolicyType": "fixed",
      "fs.s3.consistent.retryPeriodSeconds": "15",
      "fs.s3.consistent.retryCount": "20",
      "fs.s3.enableServerSideEncryption": "true",
      "fs.s3.consistent": "false"
    }
}

17/02/15 19:15:41 INFO AmazonHttpClient: Unable to execute HTTP request: randomBucket.s3.amazonaws.com:443 failed to respond
org.apache.http.NoHttpResponseException: randomBucket.s3.amazonaws.com:443 failed to respond

顺便说一句，我在hdfs中生成了数百万个小文件，并尝试了同样的工作，在一个小时内就完成了。这让我觉得它是s3特定的。还有，我用的是s3a，不是普通的S3。

Answer 1

如果您使用的是amazon EMR，那么您需要使用s3://URL；s3a://s是用于ASF版本的。

一个大问题是在s3中列出目录树需要多长时间，尤其是递归树遍历。spark代码假设它是一个快速的文件系统，其中列出dirs和声明文件的成本很低，而实际上每个操作需要1-4个HTTPS请求，即使在重用的HTTP/1.1连接上，这也是有害的。它可能很慢，你可以在日志中看到停顿。

这真正的伤害是，它是前面的分区，在那里发生了很多延迟，所以这是序列化的工作被带到膝盖上。