使用pyspark在lakehouse中遍历文件夹 需要使用 mssparkutils

当你需要在Apache Spark环境下处理大型数据湖（Lakehouse），如Azure Data Lake Storage，且希望在Pandas-like API（类似Python的pandas库）体验下进行文件夹遍历，`mssparkutils` 就派上用场了。`mssparkutils` 提供了一种更友好的方式来处理Spark DataFrames，尤其是在数据读取和操作方面。

使用`mssparkutils` 在Lakehouse中遍历文件夹并读取数据，你可以这样做：

from mssparkutils.fs.azure import AzureFileSystem
from pyspark.sql.functions import col, array, struct

# 初始化Azure FileSystem
adls = AzureFileSystem()

# 指定数据目录
data_path = 'path/to/your/directory'

# 获取目录内容（假设文件名是关键列）
directory_files = adls.ls(data_path)

# 创建DataFrame结构，将文件路径转化为struct数组
file_df = spark.createDataFrame(directory_files, schema=[("filename", StringType())])
file_array = file_df.withColumn('files', array(struct(col('filename')))).select('files').collect()

# 现在你可以遍历这个array，逐个读取文件
for file_tuple in file_array:
    filename = file_tuple[0]
    # 加载单个文件到DataFrame
    df = spark.read.format('csv') \
        .option('header', True) \
        .load(adls.path(f"{data_path}/{filename}"))
    
    # 进行处理...

在这个例子中，首先通过`ls()`函数获取目录下的所有文件名，然后将它们转换为一个结构化的数组。之后，你可以迭代这个数组，每次从中取出一个文件名，然后用它来读取和处理对应的数据。