如何在Spark Scala的S3中读取多个目录? [英] How to read multiple directories in s3 in spark Scala?

问题描述

我在s3中有以下格式的目录，

I have directories in s3 in following format,

 <base-directory>/users/users=20180303/hour=0/<parquet files>
 <base-directory>/users/users=20180303/hour=1/<parquet files>
 ....
 <base-directory>/users/users=20180302/hour=<0 to 23>/<parquet files>
 <base-directory>/users/users=20180301/hour=<0 to 23>/<parquet files>
 ....
 <base-directory>/users/users=20180228/hour=<0 to 23>/<parquet files>

基本上，我的每日目录中都有每小时的子目录.

Basically I have hourly subdirectories in daily directories.

现在我要处理最近30天的镶木地板文件.

Now I want to process parquet files the last 30 days.

我已尝试关注

 val df = sqlContext.read.option("header", "true")
    .parquet(<base-directory> + File.separator + "users" + File.separator)
    .where(col("users").between(startDate, endDate))

其中endDate和startDate间隔30天，格式为yyyymmdd.

where endDate and startDate are separated by 30 days and in yyyymmdd format.

以上解决方案未提供正确的目录子集.我在做什么错了?

Above solution is not giving correct subset of directories. What am I doing wrong ?

推荐答案

其中函数用于数据帧中的过滤行.您将其用于从 s3 读取 parquet 文件.所以整个概念是错误的.

where function is used in filtering rows in dataframe. And you are using it for read parquet files from s3. So the whole concept is wrong.

相反，您可以在startDate和endDate之间创建路径数组，并将其传递给 sqlContext read api .

Instead you can create an array of paths between startDate and endDate and pass it to sqlContext read api.

以编程方式讲，您可以执行以下操作(它们只是伪代码)

Programmatically speaking, you can do something like below (they are just pseudo code)

val listBuffer = new ListBuffer[String]
for(date <- startDate to endDate)
  listBuffer.append(<base-directory> + File.separator + "users" + File.separator+"users="+date)

val df = sqlContext.read.option("header", "true").parquet(listBuffer: _*)

这篇关于如何在Spark Scala的S3中读取多个目录?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！