按照你的规模，优雅的解决方案将是一个延伸。
我建议不要使用 sc.textFile("s3n://mybucket/path/txt/*/*/*/*.txt.gz") 因为这需要永远。您可以使用aws distcp或类似的方法将文件移动到hdfs中。一旦进入hdfs，spark就会以任何适合您的方式快速地接收信息。
请注意，大多数这些过程都需要某种类型的文件列表，因此您需要以某种方式生成它。对于20 mil文件，创建文件列表将是一个瓶颈。我建议每次将文件上传到s3时，创建一个附加了文件路径的文件。
同样的输出，放入hdfs，然后移到s3（尽管直接拷贝可能同样有效）。

你试过类似sc.wholetextfiles的东西吗？
它创建一个rdd，其中键是文件名，值是整个文件的字节数组。然后可以将其Map，使键为文件日期，然后再Mapgroupbykey？
http://spark.apache.org/docs/latest/programming-guide.html

你可以用这个
首先，您可以获得s3路径的缓冲区/列表：

import scala.collection.JavaConverters._
import java.util.ArrayList
import com.amazonaws.services.s3.AmazonS3Client
import com.amazonaws.services.s3.model.ObjectListing
import com.amazonaws.services.s3.model.S3ObjectSummary
import com.amazonaws.services.s3.model.ListObjectsRequest

def listFiles(s3_bucket:String, base_prefix : String) = {
    var files = new ArrayList[String]

    //S3 Client and List Object Request
    var s3Client = new AmazonS3Client();
    var objectListing: ObjectListing = null;
    var listObjectsRequest = new ListObjectsRequest();

    //Your S3 Bucket
    listObjectsRequest.setBucketName(s3_bucket)

    //Your Folder path or Prefix
    listObjectsRequest.setPrefix(base_prefix)

    //Adding s3:// to the paths and adding to a list
    do {
      objectListing = s3Client.listObjects(listObjectsRequest);
      for (objectSummary <- objectListing.getObjectSummaries().asScala) {
        files.add("s3://" + s3_bucket + "/" + objectSummary.getKey());
      }
      listObjectsRequest.setMarker(objectListing.getNextMarker());
    } while (objectListing.isTruncated());

    //Removing Base Directory Name
    files.remove(0)

    //Creating a Scala List for same
    files.asScala
  }

现在将这个list对象传递给下面的代码段，注意：sc是sqlcontext的一个对象

var df: DataFrame = null;
  for (file <- files) {
    val fileDf= sc.textFile(file)
    if (df!= null) {
      df= df.unionAll(fileDf)
    } else {
      df= fileDf
    }
  }

现在你得到了最终的统一rdd，即df
可选，您还可以在一个bigrdd中重新划分它

val files = sc.textFile(filename, 1).repartition(1)

重新分区始终有效：d