合并来自 Spark 中多个目录的日志

Question

我根据日志文件的创建日期将日志文件放入不同的目录。

例如

> /mypath/2017/01/20/... 
.
.
.
> /mypath/2017/02/13/...
> /mypath/2017/02/14/...

我想使用 pyspark 将所有这些日志文件合并到一个 rdd 中，以便我可以对这个主文件进行聚合。

到目前为止，我已经采用了称为sqlContext的各个目录，并使用Union来连接特定日期的所有日志文件。

DF1 = (sqlContext.read.schema(schema).json("/mypath/2017/02/13")).union(sqlContext.read.schema(schema).json("/mypath/2017/02/14"))

有没有一种简单的方法可以通过指定日期范围内的日志文件来获取主rdd？(即由2017/01/20至2017/02/14(

我对火花很陌生，如果我在任何一步都错了，请纠正我。

Answer 1

如果您坚持使用 sqlContext，那么一个简单的解决方案是定义一种方法，该方法将列出输入目录中的所有文件

case class FileWithDate(basePath: String, year: Int, month: Int, day: Int) {
 def path = s"${basePath}/${year}/${month}/${day}"
}
def listFileSources() : List[FileWithDate] = ??? // implement here

如果要合并来自源的所有数据帧，可以这样做：

// create an empty dataframe with the strucutre for the json
val files = listSources()
val allDFs = files.foldLeft(emptyDF){case (df, f) => df.union(sqlContext.read.schema(schema).json(f.path))}

如果要按日期过滤输入文件，那将很容易。像这样的东西

files.filter(_.year == 2016 && (_.month >=2 || _.month <=3))

另一种解决方案是用年、月、日扩充数据帧(放置额外的列(，并在新数据帧上执行所有业务逻辑