scala - 按日期周期从 Spark 中的 S3 读取多个文件

Question

描述

我有一个应用程序，它将数据发送到 AWS Kinesis Firehose，然后将数据写入我的 S3 存储桶。 Firehose 使用“yyyy/MM/dd/HH”格式写入文件。

就像这个示例 S3 路径一样:

s3://mybucket/2016/07/29/12

现在我有一个用Scala编写的Spark应用程序，我需要读取特定时间段的数据。我有开始日期和结束日期。数据采用 JSON 格式，这就是我使用 sqlContext.read.json() 而不是 sc.textFile() 的原因。

如何快速高效地读取数据？

我尝试了什么？

1. 通配符 - 我可以选择特定日期的所有小时或特定月份的所有日期的数据，例如:

   val df = sqlContext.read.json("s3://mybucket/2016/07/29/*")
   val df = sqlContext.read.json("s3://mybucket/2016/07/*/*")
   

   但是，如果我必须读取几天的日期期间的数据，例如 2016-07-29 - 2016-07-30，我就不能以同样的方式使用通配符方法。

   这引出了我的下一点......

2. 使用多个路径或 CSV 目录，如 this 中的 samthebest 所示解决方案。似乎用逗号分隔目录仅适用于 sc.textFile() 而不是 sqlContext.read.json()。

3. Union - cloud 上一个链接的第二个解决方案建议单独读取每个目录，然后将它们合并在一起。尽管他建议联合 RDD，但也可以选择联合 DataFrame。如果我手动生成给定日期期间的日期字符串，那么我可能会创建一个不存在的路径，而不是忽略它，整个读取会失败。相反，我可以使用 AWS SDK 并使用 AmazonS3Client 中的函数 listObjects 来获取所有 key ，就像上一个链接中的 iMKanchwala 解决方案中一样。

   唯一的问题是我的数据不断变化。如果 read.json() 函数将所有数据作为单个参数获取，它会读取所有必要的数据，并且足够智能，可以从数据中推断出 json 模式。如果我分别读取 2 个目录并且它们的架构不匹配，那么我认为合并这两个数据帧会成为一个问题。

4. Glob(?) 语法 - This nhahtdh 的解决方案比选项 1 和 2 稍好一些，因为它们提供了更详细地指定日期和目录的选项，并且作为单个选项“path”，因此它也适用于 read.json()。

   但是，一个常见的问题又出现了，即丢失的目录。假设我想要从 20.07 到 30.07 的所有数据，我可以这样声明:

   val df = sqlContext.read.json("s3://mybucket/2016/07/[20-30]/*")
   

   但是，如果我丢失了 7 月 25 日的数据，则路径 ..16/07/25/ 不存在，整个函数就会失败。

显然，当请求的时间段是例如 2015 年 11 月 25 日至 2016 年 2 月 12 日时，事情会变得更加困难，那么我需要以编程方式(在我的 Scala 脚本中)创建一个如下所示的字符串路径:

"s3://mybucket/{2015/11/[25-30],2015/12/*,2016/01/*,2016/02/[01-12]}/*"

通过创建它，我需要以某种方式确保这些 25-30 和 01-12 间隔都有相应的路径，如果缺少一个，它会再次失败。 (星号幸运地处理丢失的目录，因为它读取存在的所有内容)

如何一次从单个目录路径读取所有必要的数据，而不可能因为某个日期间隔之间缺少目录而失败？

Answer 1

有一个更简单的解决方案。如果你看DataFrameReader API您会注意到有一个 .json(paths: String*) 方法。只需构建一个您想要的路径的集合，根据您的喜好，使用 not 的全局变量，然后调用该方法，例如，

val paths: Seq[String] = ...
val df = sqlContext.read.json(paths: _*)