从文件系统有条件加载分区

3zwtqj6y 于 2021-05-27 发布在 Spark

关注(0)|答案(1)|浏览(366)

我知道在pysparks中有关于通配符的问题 .load() -像这里或这里这样运作。不管怎样，我找到的问题/答案中没有一个涉及到我的变化。

上下文

在pyspark中，我想直接从hdfs加载文件，因为我必须使用databricks avro library for spark 2.3.x。我是这样做的：

partition_stamp = "202104"

df = spark.read.format("com.databricks.spark.avro") \
        .load(f"/path/partition={partition_stamp}*") \
        .select("...")

如您所见，分区是从格式中的时间戳派生的 yyyyMMdd .

问题

目前我只得到2021年4月使用的所有分区( partition_stamp = "202104" ). 但是，我需要从2021年4月开始的所有分区。
用伪代码编写，我需要一个类似的解决方案：

.load(f"/path/partition >= {partition_stamp}*")

因为实际上有几百个分区，所以以任何需要硬编码的方式来做是没有用的。
所以我的问题是：是否有一个函数用于条件文件加载？

avro apache-spark pyspark spark-avro

来源：https://stackoverflow.com/questions/63234234/conditional-loading-of-partitions-from-file-system

1条答案

按热度按时间

8e2ybdfx1#

据我所知，只有以下选项可以动态处理 .load() -功能：


* :  Wildcard for any character or sequence of characters until the end of the line or a new sub-directory ('/') -> (/path/20200*)

[1-3]: Regex-like inclusion of a defined character-range -> (/path/20200[1-3]/...)
{1,2,3}: Set-like inclusion of a defined set of characters -> (/path/20200{1,2,3}/...)

因此，回答我的问题：没有用于条件文件加载的内置函数。
不管怎样，我想为您提供我的解决方案：

import pandas as pd # Utilize pandas date-functions

partition_stamp = ",".join((set(
                        str(_range.year) + "{:02}".format(_range.month) 
                        for _range in pd.date_range(start=start_date, end=end_date, freq='D')
                 )))

df = spark.read.format("com.databricks.spark.avro") \
        .load(f"/path/partition={{{partition_stamp}}}*") \
        .select("...")

这样就减少了对时间戳格式的限制 yyyyMM 为给定的开始和结束日期以及基于字符串的 .load() 仍然可用。

赞(0）回复(0）举报 2021-05-27

我来回答

从文件系统有条件加载分区

上下文

问题

1条答案

相关问题

热门标签

最新问答