如何将Spark DataFrame格式化为带颜色的Excel并写入Azure存储

核心思路 #

Spark原生不支持带格式的Excel输出，必须转成Pandas DataFrame后，用Python的Excel处理库（如openpyxl）设置样式，再处理存储（包括Blob服务）。你之前用create_blob_from_text失败是因为Excel是二进制格式，不是纯文本，不能用文本写入方法。

具体实现步骤 #

1. 转换Spark DataFrame到Pandas DataFrame #

先将Spark的结构化数据转为Pandas DataFrame，大数据量时建议先过滤、采样或分区处理，避免内存溢出：

# 假设spark_df是你的Spark结构化数据表
pandas_df = spark_df.toPandas()

2. 用openpyxl创建带样式的Excel #

通过openpyxl定义并应用所需样式：

from openpyxl import Workbook
from openpyxl.styles import PatternFill
from openpyxl.utils.dataframe import dataframe_to_rows

# 定义填充样式
required_fill = PatternFill(start_color="FFA500", end_color="FFA500", fill_type="solid")  # 必填列橙色
optional_fill = PatternFill(start_color="FFFF00", end_color="FFFF00", fill_type="solid")  # 可选列黄色
missing_row_fill = PatternFill(start_color="FF0000", end_color="FF0000", fill_type="solid")  # 缺失数据行红色

# 创建工作簿和工作表
wb = Workbook()
ws = wb.active

# 将Pandas DataFrame写入工作表
for r in dataframe_to_rows(pandas_df, index=False, header=True):
    ws.append(r)

# 标记必填列和可选列（自行替换为你的列名列表）
required_cols = ["user_id", "order_no"]
optional_cols = ["user_phone", "delivery_addr"]

# 处理表头样式
header_row = ws[1]
for cell in header_row:
    if cell.value in required_cols:
        cell.fill = required_fill
    elif cell.value in optional_cols:
        cell.fill = optional_fill

# 标记存在数据缺失的行
for row in ws.iter_rows(min_row=2):
    has_missing = any(cell.value is None or cell.value == "" for cell in row)
    if has_missing:
        for cell in row:
            cell.fill = missing_row_fill

3. 上传到Blob存储 #

使用二进制流方式上传，而非文本写入：

from azure.storage.blob import BlockBlobService
import io

# 将工作簿保存到内存字节流
output = io.BytesIO()
wb.save(output)
output.seek(0)  # 重置流指针到起始位置

# 初始化Blob服务并上传
block_blob_service = BlockBlobService(account_name="你的存储账户名", account_key="你的存储密钥")
block_blob_service.create_blob_from_bytes(
    container_name="你的容器名",
    blob_name="带样式输出.xlsx",
    blob=output.getvalue()
)

注意事项 #

• 若数据量极大，直接转Pandas可能触发内存不足，建议先通过Spark做数据过滤或分区处理，再分批转换合并。
• 确保安装依赖：pip install openpyxl pandas azure-storage-blob pyspark
• 列名需与实际数据严格匹配，避免样式应用错误。

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