PySpark实现：提取Azure文件夹中每个CSV文件的首行

要实现只读取CSV文件首行、避免全量加载的需求，可以借助Hadoop的FileSystem API直接读取每个文件的第一行，结合PySpark生成目标DataFrame，步骤如下：

1. 配置Azure存储访问权限 #

首先需要配置Spark连接Azure Blob Storage或ADLS Gen2的权限，以ADLS Gen2为例：

# 替换为你的存储账户名和访问密钥
spark.conf.set(
    "fs.azure.account.key.your-storage-account.dfs.core.windows.net",
    "your-access-key-here"
)
# 若使用SAS令牌，替换为以下配置
# spark.conf.set(
#     "fs.azure.sas.your-container.your-storage-account.dfs.core.windows.net",
#     "your-sas-token-here"
# )

2. 获取目标文件夹下的所有CSV文件路径 #

通过Hadoop FileSystem API遍历目标文件夹，筛选出CSV文件：

from pyspark.sql.types import StringType
import org.apache.hadoop.fs as fs

# 替换为Azure上的文件夹路径，格式如abfss://<container>@<storage-account>.dfs.core.windows.net/<folder>
folder_path = "abfss://your-container@your-storage-account.dfs.core.windows.net/csv-folder"
file_system = fs.FileSystem.get(spark.sparkContext._jsc.hadoopConfiguration())

# 递归=False表示只读取当前文件夹，如需递归子文件夹设为True
file_paths = [
    f.getPath().toString() 
    for f in file_system.listFiles(fs.Path(folder_path), recursive=False) 
    if f.getPath().getName().endswith(".csv")
]

3. 定义UDF读取文件首行 #

创建自定义函数，通过FileSystem API读取单个文件的第一行，避免全量加载：

def extract_first_line(file_path):
    file = file_system.open(fs.Path(file_path))
    # 若CSV使用其他编码，替换为对应编码（如gbk）
    first_line = file.readline().decode("utf-8").strip()
    file.close()
    return first_line

# 注册UDF
get_first_line_udf = spark.udf.register("extract_first_line", extract_first_line, StringType())

4. 生成目标DataFrame #

将文件路径转为DataFrame，再添加文件名和首行内容列：

from pyspark.sql.functions import input_file_name

# 创建文件路径DataFrame
df = spark.createDataFrame(file_paths, StringType()).toDF("file_path")

# 提取文件名、首行内容
result_df = df.withColumn(
    "filename", 
    input_file_name().substr(input_file_name().lastIndexOf("/") + 1, 1000)
).withColumn(
    "firstrow", 
    get_first_line_udf("file_path")
)

# 查看结果
result_df.show(truncate=False)

注意事项 #

• 若CSV文件使用非UTF-8编码，需调整decode方法中的编码参数（如gbk）；
• 如需递归读取子文件夹中的CSV，将listFiles的recursive参数设为True；
• 该方法通过直接读取文件首字节实现，不会加载整个文件，适合处理大体积CSV。

内容的提问来源于stack exchange，提问作者Graham C