python - 读取 \yyyy\mm\dd\HH\MM\ 范围内的文件
问题描述
我有一个 PySpark 应用程序,它需要从 Azure blob 存储帐户读取文件,其中文件每 5 分钟以这种格式分区到文件夹中:
\Root\yyyy\mm\dd\HH\MM\files.csv
我有一个每小时运行一次的进程,并且想要处理自上次运行以来的所有文件(如果错过运行可能会超过一个小时)。我管理了一个高水位线,它告诉我最后处理的文件夹时间。
在文件内部还有一个与路径日期时间匹配的日期时间字段(第二个更详细)。
请注意,我无法将文件夹结构更改为 Sparks 首选分区方法 year=yyyy\month=mm 等。
我写了这个函数:
from datetime import datetime
def folderDateTimeRange(startDateTime, endDateTime, levels=5):
if startDateTime.year != endDateTime.year:
return '/{*}' * levels
elif startDateTime.month != endDateTime.month:
return datetime.strftime(startDateTime, '%Y') + '/{*}' * (levels - 1)
elif startDateTime.day != endDateTime.day:
return datetime.strftime(startDateTime, '%Y/%m') + '/{*}' * (levels - 2)
elif startDateTime.hour != endDateTime.hour:
return datetime.strftime(startDateTime, '%Y/%m/%d') + '/{*}' * (levels - 3)
else:
return ""
这限制了在大多数情况下读取的文件夹数量。我仍然需要过滤由传递给函数的相同开始和结束时间读取的数据,因为第二天 23:00 到 01:00 将在日期和小时部分返回 {*} - 因此我认为这可能更有效率。
在最糟糕的示例中,您传入 start = 2018-12-31 22:00:00 和 end = 2019-01-01 01:00:00 - 这会导致读取所有年份的所有数据。
我对 glob 的了解有限 - 但是否可以传递一个范围而不是 {*}?
解决方案
是的,您可以使用花括号返回项目列表,也可以使用正则表达式。
Check here: Read range of files in pySpark and here: pyspark select subset of files using regex/glob from s3 (I am not sure how much Azure and S3 differ but my assumption is PySpark can abstract this away; correct me if I'm wrong.)
You can also minimize your 'waste' of reading files by generating a few paths and sending them instead of just a single path (this ensures you won't have the same pitfall of reading two years of data if as you cross from one year to the next.)
For fun I wrote a little code with some test stuff at the bottom, you can probably return these lists and get what you want:
from datetime import datetime as dt
from datetime import timedelta
from collections import defaultdict
# \Root\yyyy\mm\dd\HH\MM\files.csv
def folderDateTimeRange(start, end, levels=5):
start_iter = start
paths = defaultdict(lambda: defaultdict(lambda: defaultdict(lambda: defaultdict(list))))
while start_iter < end:
paths[start_iter.year][start_iter.month][start_iter.day][start_iter.hour].append(start_iter.minute)
start_iter += timedelta(minutes=5)
ret_paths = []
for year, v1 in paths.items():
path = '{}\\'.format(year)
for month, v2 in v1.items():
path += '{}\\'.format(month)
for day, v3 in v2.items():
path += '{}\\'.format(day)
path += '{{{}}}\\{{*}}'.format(','.join([str(_) for _ in v3.keys()]))
ret_paths.append(path)
return ret_paths
def test(a, b):
res = folderDateTimeRange(a, b)
for r in res:
print(r)
print('---')
test(dt(2018, 1, 1), dt(2018, 1, 2))
test(dt(2018, 12, 31), dt(2019, 1, 2))