我想从S3将多个实木复合地板文件读取到一个数据帧中。目前,我正在使用以下方法执行此操作:
files = ['s3a://dev/2017/01/03/data.parquet', 's3a://dev/2017/01/02/data.parquet'] df = session.read.parquet(*files)
如果所有文件都存在于S3上,则此方法有效,但是当列表中的某些文件不存在时,我想请求将文件列表加载到数据帧中而不会中断。换句话说,我希望sparkSql可以将它找到的尽可能多的文件加载到数据帧中,并返回此结果而不会抱怨。这可能吗?
是的,如果您将指定输入的方法更改为hadoop glob模式,则是可能的,例如:
files = 's3a://dev/2017/01/{02,03}/data.parquet' df = session.read.parquet(files)
您可以在Hadoop javadoc中阅读有关模式的更多信息。
但是,在我看来,这不是按时间(按您的情况按天)划分的数据处理的优雅方式。如果您能够这样重命名目录:
s3a://dev/2017/01/03/data.parquet
-> s3a://dev/day=2017-01-03/data.parquet
s3a://dev/2017/01/02/data.parquet
-> s3a://dev/day=2017-01-02/data.parquet
那么您可以利用Spark分区模式并通过以下方式读取数据:
session.read.parquet('s3a://dev/') \ .where(col('day').between('2017-01-02', '2017-01-03')
这种方式也将省略空目录/不存在的目录。Additionall列day
将出现在您的数据框中(在spark <2.1.0中为字符串,在spark> = 2.1.0中为datetime),因此您将知道每个记录位于哪个目录中。