组合来自 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）

刚接触spark，如有不妥请指正

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))

另一个解决方案是用年、月、日来扩充您的数据框（放置额外的列），并在新数据框上执行所有业务逻辑

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

Combining the logs from multiple directories in Spark

scala

apache-spark

pyspark

pyspark-sql

databricks