如何用Python/PySpark获取Blob存储中指定日期的文件名列表

方法一：Python原生实现（依赖azure-storage-blob SDK） #

前置准备 #

先安装所需依赖：

pip install azure-storage-blob

实现代码 #

from azure.storage.blob import BlobServiceClient
from datetime import date

# 配置参数
STORAGE_ACCOUNT_CONNECTION_STRING = "你的存储账户连接字符串"
CONTAINER_NAME = "目标容器名"
TARGET_DATE = date(2022, 9, 12)  # 目标日期，格式为年/月/日

# 初始化Blob服务客户端
blob_service_client = BlobServiceClient.from_connection_string(STORAGE_ACCOUNT_CONNECTION_STRING)
container_client = blob_service_client.get_container_client(CONTAINER_NAME)

# 遍历并过滤Blob
target_files = []
for blob in container_client.list_blobs():
    # *注意：Blob的last_modified是UTC时间*，需根据需求调整时区
    blob_modified_date = blob.last_modified.date()
    if blob_modified_date == TARGET_DATE:
        target_files.append(blob.name)

# 输出结果
print("特定日期上传的文件列表：")
print(target_files)

注意事项 #

• 若目标日期为本地时区而非UTC，需对blob.last_modified进行时区转换后再提取日期。
• 容器内文件数量极大时，可使用list_blobs()的分页参数优化遍历性能。

方法二：PySpark实现 #

前置配置 #

确保Spark环境已配置Azure Blob存储的访问权限，可通过Spark配置项设置存储账户密钥或SAS令牌。

实现代码 #

from pyspark.sql import SparkSession
from pyspark.sql.functions import date, col

# 初始化SparkSession
spark = SparkSession.builder \
    .appName("BlobDateFilter") \
    .config("fs.azure.account.key.<存储账户名>.blob.core.windows.net", "<存储账户密钥>") \
    .getOrCreate()

# 目标Blob路径（格式：wasbs://<容器名>@<存储账户名>.blob.core.windows.net/）
BLOB_PATH = "wasbs://your-container@your-account.blob.core.windows.net/"
TARGET_DATE_STR = "2022-09-12"

# 读取文件元数据（binaryFile格式返回文件路径、修改时间等元数据）
df = spark.read.format("binaryFile").load(BLOB_PATH)

# 过滤特定日期的文件并提取文件名
target_files_df = df.filter(date(col("last_modified")) == TARGET_DATE_STR) \
    .select(col("path").alias("file_name"))

# 转为列表输出
target_files = [row.file_name.split("/")[-1] for row in target_files_df.collect()]
print("特定日期上传的文件列表：")
print(target_files)

注意事项 #

• 注意：last_modified字段默认是UTC时间，若目标日期为本地时区，需通过to_timestamp和to_utc_timestamp进行时区转换。
• 若仅需文件元数据而非二进制内容，可改用spark.sparkContext.wholeTextFiles()减少资源占用。

内容的提问来源于stack exchange，提问作者Swati B