开发者社区> 问答> 正文

如何使用EMR上的spark有效地读取/解析s3文件夹中.gz文件的负载

小六码奴 2019-04-22 15:45:07 274

我正在尝试通过在EMR上执行的spark应用程序读取s3上目录中的所有文件。

数据以典型格式存储,如“s3a://Some/path/yyyy/mm/dd/hh/blah.gz”

如果我使用深度嵌套的通配符(例如“s3a:// SomeBucket / SomeFolder / / / / *。gz”),性能非常糟糕,需要大约40分钟才能读取几万个小的gzip压缩文件。
我的另外两种方法,我的研究告诉我,它的性能要高得多。

使用hadoop.fs库(2.8.5)我尝试读取我提供的每个文件路径。

private def getEventDataHadoop(

eventsFilePaths: RDD[String]

)(implicit sqlContext: SQLContext): Try[RDD[String]] =

Try(
  {
    val conf = sqlContext.sparkContext.hadoopConfiguration

    eventsFilePaths.map(eventsFilePath => {
      val p                            = new Path(eventsFilePath)
      val fs                           = p.getFileSystem(conf)
      val eventData: FSDataInputStream = fs.open(p)
      IOUtils.toString(eventData)
    })
  }
)

这些文件路径由以下代码生成:

private[disneystreaming] def generateInputBucketPaths(

s3Protocol: String,
bucketName: String,
service: String,
region: String,
yearsMonths: Map[String, Set[String]]

): Try[Set[String]] =

Try(
  {
    val days                         = 1 to 31
    val hours                        = 0 to 23
    val dateFormatter: Int => String = buildDateFormat("00")

    yearsMonths.flatMap { yearMonth: (String, Set[String]) =>
      for {
        month: String <- yearMonth._2
        day: Int      <- days
        hour: Int     <- hours
      } yield
        s"$s3Protocol$bucketName/$service/$region/${dateFormatter(yearMonth._1.toInt)}/${dateFormatter(month.toInt)}/" +
          s"${dateFormatter(day)}/${dateFormatter(hour)}/*.gz"
    }.toSet
  }
)

hadoop.fs代码失败,因为Path类不可序列化。我想不出怎么能解决这个问题。

所以这导致我使用AmazonS3Client的另一种方法,我只是要求客户端给我文件夹(或前缀)中的所有文件路径,然后将文件解析为字符串,由于它们被压缩可能会失败:

private def getEventDataS3(bucketName: String, prefix: String)(

implicit sqlContext: SQLContext

): Try[RDD[String]] =

Try(
  {
    import com.amazonaws.services.s3._, model._
    import scala.collection.JavaConverters._

    val request = new ListObjectsRequest()
    request.setBucketName(bucketName)
    request.setPrefix(prefix)
    request.setMaxKeys(Integer.MAX_VALUE)
    val s3 = new AmazonS3Client(new ProfileCredentialsProvider("default"))

    val objs: ObjectListing = s3.listObjects(request) // Note that this method returns truncated data if longer than the "pageLength" above. You might need to deal with that.
    sqlContext.sparkContext
      .parallelize(objs.getObjectSummaries.asScala.map(_.getKey).toList)
      .flatMap { key =>
        Source
          .fromInputStream(s3.getObject(bucketName, key).getObjectContent: InputStream)
          .getLines()
      }
  }
)

此代码产生null异常,因为配置文件不能为null(“java.lang.IllegalArgumentException:配置文件不能为null”)。请记住,此代码在AWS中的EMR上运行,因此如何提供所需的凭据?其他人如何使用此客户端在EMR上运行spark作业?

存储 分布式计算 Spark
分享到
取消 提交回答
全部回答(1)
  • 小六码奴
    2019-07-17 23:33:58

    Path在后来的Hadoop版本中是可序列化的,因为它可以在Spark RDD中使用。在此之前,将路径转换为URI,对其进行编组,并在闭包内从该URI创建新路径。

    0 0
+ 订阅

大数据计算实践乐园,近距离学习前沿技术

推荐文章
相似问题
推荐课程