如何使用Azure Databricks PySpark读写ADLS Gen2多Sheet的Excel数据

前置依赖配置 #

• Databricks集群需预先安装com.crealytics:spark-excel Maven依赖，版本需要和集群的Scala、Spark版本匹配，例如Spark 3.3 + Scala 2.12环境对应版本为com.crealytics:spark-excel_2.12:3.3.1_0.18.7
• 安装openpyxl库用于读取工作表名称，可在notebook中执行%pip install openpyxl完成安装
• 提前完成ADLS Gen2访问权限配置，可通过挂载、SAS令牌或服务主体验证的方式让集群获得对应路径的读写权限

完整实现代码 #

# 引入所需依赖
from pyspark.sql.functions import lit
import openpyxl

# 1. 定义输入输出路径，替换为你自己的ADLS路径
input_excel_path = "abfss://<容器名>@<存储账号名>.dfs.core.windows.net/<Excel文件存放路径>.xlsx"
output_path = "abfss://<容器名>@<存储账号名>.dfs.core.windows.net/<合并后数据输出路径>"

# 2. 读取Excel文件中所有工作表名称
wb = openpyxl.load_workbook(input_excel_path, read_only=True)
sheet_names = wb.sheetnames
wb.close()

# 3. 预定义表Schema，避免自动类型推断出错
custom_schema = "Id INT, Name STRING"

# 4. 循环读取每个工作表数据，新增存储工作表名称的字段
df_list = []
for sheet_name in sheet_names:
    single_sheet_df = spark.read.format("com.crealytics.spark.excel") \
        .option("header", "true") \
        .option("dataAddress", f"'{sheet_name}'!A1") \
        .schema(custom_schema) \
        .load(input_excel_path) \
        .withColumn("sheet_name", lit(sheet_name))
    df_list.append(single_sheet_df)

# 5. 合并所有工作表的数据
merged_all_df = df_list[0].unionByName(*df_list[1:])

# 6. 将合并后的数据写入ADLS Gen2指定路径，可根据需求更换为parquet、csv等格式
merged_all_df.write.format("delta") \
    .mode("overwrite") \
    .save(output_path)

注意事项 #

• 若处理的Excel文件体积过大，可将读取工作表名称的逻辑替换为分布式执行方案，避免Driver节点负载过高
• 写入时可根据业务需求修改mode参数：overwrite为覆盖已有数据写入，append为在已有数据基础上追加写入
• 若需要保留Excel中的空值行，可在读取Excel的配置中添加.option("treatEmptyValuesAsNulls", "true")

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