从 S3 加载 spark DF，多个文件。这些方法中哪一种最好？

Posted 2023-04-15

【中文标题】从 S3 加载 spark DF，多个文件。这些方法中哪一种最好？

【英文标题】：Loading a spark DF from S3, multiple files. Which of these approaches is best?

【发布时间】：2018-11-30 17:13:24

【问题描述】：

我有一个 s3 存储桶，其中包含 Athena 底层的分区数据。使用 Athena，我看到我的表中有 1040 亿行。这是大约2年的数据。 我们就叫它big_table吧。

分区是按天，按小时，所以每天 07-12-2018-00,01,02 ... 24。 Athena 字段是partition_datetime。

在我的用例中，我只需要 1 个月的数据，大约 4 亿行。

所以问题出现了 - 直接从以下位置加载： 1. 文件

spark.load(['s3://my_bucket/my_schema/my_table_directory/07-01-2018-00/file.snappy.parquet',\
         's3://my_bucket/my_schema/my_table_directory/07-01-2018-01/file.snappy.parquet' ],\
         .
         .
         .
         's3://my_bucket/my_schema/my_table_directory/07-31-2018-23/file.snappy.parquet'])

或 2. 通过 pyspark 使用 SQL

df = spark.read.parquet('s3://my_bucket/my_schema/my_table_directory')
df = df.registerTempTable('tmp')
df = spark.sql("select * from my_schema.my_table_directory where partition_datetime >= '07-01-2018-00' and partition_datetime < '08-01-2018-00'")

我认为 #1 更有效，因为我们只引入了相关时期的数据。

2 对我来说似乎效率低下，因为必须遍历整个 1040 亿行（或更准确地说是 partition_datetime 字段）才能满足 SELECT。我被告知这确实不是问题，因为执行延迟，并且从来没有一个包含所有 1040 亿行的 df。我仍然说在某些时候必须由 SELECT 访问每个分区，因此选项 1 更有效。

我对这方面的其他意见感兴趣。请插话

【参考方案1】：

您所说的可能是真的，但它效率不高，因为它永远不会扩展。如果你想要三个月的数据，你不能在你的加载命令中指定 90 行代码。当涉及到大数据时，这不是一个好主意。您始终可以使用独立的 spark 或 YARN 集群对这么大的数据集执行操作。

【参考方案2】：

您可以在路径中使用通配符来仅加载给定范围内的文件。

spark.read.parquet('s3://my_bucket/my_schema/my_table_directory/07-01,02,03-2018-*/')

或

spark.read.parquet('s3://my_bucket/my_schema/my_table_directory/07-*-2018-*/')

【参考方案3】：

汤姆，你是对的。 ＃1更有效，也是这样做的方式。但是，您可以创建要读取的文件列表的集合，然后让 spark 仅读取这些文件。

此blog 可能对您的情况有所帮助。

【讨论】：

谁否认了我的回应，请您用您的经验告诉我们，这有助于整个社区，而不是默默的否定？

使用 Manoj 回复的通配符是被广泛提及的解决方案。但是我提到的解决方案适用于当您有太多文件时，在程序开始时获取文件列表需要太多时间，或者如果您遇到内存不足的问题。很少有其他帖子建议做同样的事情。当然，这取决于OP面临的实际情况。 ***.com/questions/34662953/…garrens.com/blog/2017/11/04/…