如何在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 ?

推荐答案

where 函数用于dataframe 中的过滤行.您正在使用它从 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屋！