Azure Data Factory自动化.xlsx转.csv并同步至Azure SQL DB方案咨询

刚接触Azure就能搭出手动流程已经很棒了！你提到的Python脚本和Databricks笔记本都是可行的方案，另外还有Azure Function这种事件驱动的选项，我来帮你拆解每个方案的实现思路、优缺点和实操细节：

方案1：ADF管道内直接嵌入Python脚本活动 #

适合轻量文件转换、不想额外维护服务的场景，直接在现有ADF管道里扩展，不需要新增独立服务。

实现步骤： #

1. 确保你有自托管集成运行时（Self-hosted IR）——因为ADF托管IR不支持自定义Python库，自托管IR可以安装pandas、openpyxl（处理XLSX）和azure-storage-blob依赖。
2. 在ADF管道中添加「Python脚本」活动，配置使用自托管IR。
3. 编写Python脚本实现从Blob读XLSX、转CSV、再写回Blob，代码示例：

import pandas as pd
from azure.storage.blob import BlobServiceClient
import os

# 用ADF参数传递关键信息（避免硬编码）
connect_str = "@pipeline().parameters.blobConnectionString"
container_name = "@pipeline().parameters.containerName"
input_file = "@pipeline().parameters.inputXlsxFileName"
output_file = input_file.replace(".xlsx", ".csv")

# 初始化Blob客户端
blob_client = BlobServiceClient.from_connection_string(connect_str).get_container_client(container_name)

# 临时文件处理（自托管IR本地目录）
temp_xlsx = "/tmp/temp_input.xlsx"
temp_csv = "/tmp/temp_output.csv"

# 下载XLSX到本地
with open(temp_xlsx, "wb") as f:
    f.write(blob_client.download_blob(input_file).readall())

# 转换为CSV
df = pd.read_excel(temp_xlsx, engine="openpyxl")
df.to_csv(temp_csv, index=False, encoding="utf-8-sig")  # 带BOM避免中文乱码

# 上传CSV回Blob
with open(temp_csv, "rb") as f:
    blob_client.upload_blob(output_file, f, overwrite=True)

# 清理临时文件
os.remove(temp_xlsx)
os.remove(temp_csv)

4. 把脚本里的连接信息、文件名做成ADF管道参数，方便后续复用和调度。

优缺点： #

• ✅ 无需额外服务，和现有ADF流程无缝整合
• ✅ 成本低，仅消耗自托管IR资源
• ❌ 处理超大文件（比如10GB+）时性能有限，依赖自托管IR的硬件配置

方案2：使用Databricks笔记本处理转换 #

适合大文件、复杂数据清洗逻辑的场景，Databricks的Spark引擎对批量数据处理更高效，也支持更复杂的Excel格式（比如多Sheet、合并单元格）。

实现步骤： #

1. 创建Azure Databricks工作区和集群，在集群的「库」中添加Spark Excel依赖（com.crealytics:spark-excel_2.12:0.13.5）。
2. 编写Databricks笔记本，实现XLSX转CSV：

# 用ADF传递的参数获取文件名
input_file = dbutils.widgets.get("inputFileName")
output_file = input_file.replace(".xlsx", ".csv")

# 配置Blob存储访问（推荐用Service Principal而非密钥）
storage_account = "<你的存储账户名>"
container = "<容器名>"
spark.conf.set(f"fs.azure.account.key.{storage_account}.blob.core.windows.net", "<存储账户密钥>")

# 读取XLSX
df = spark.read.format("com.crealytics.spark.excel") \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .option("dataAddress", "'Sheet1'!A1")  # 指定Sheet和起始单元格
    .load(f"wasbs://{container}@{storage_account}.blob.core.windows.net/{input_file}")

# 写入CSV
df.write.format("csv") \
    .option("header", "true") \
    .option("encoding", "utf-8") \
    .mode("overwrite") \
    .save(f"wasbs://{container}@{storage_account}.blob.core.windows.net/{output_file}")

3. 在ADF管道中添加「Databricks笔记本」活动，传递文件名参数，调用上述笔记本执行转换。

优缺点： #

• ✅ 支持超大文件和复杂数据处理，性能远超单节点Python
• ✅ 可扩展：后续如果需要添加数据清洗、格式校验，直接在笔记本里扩展即可
• ❌ 需要维护Databricks集群，成本略高（按需付费可降低成本）

方案3：Azure Function事件触发自动化 #

适合实时/准实时处理的场景——市场部一上传XLSX到Blob，就自动触发转换，不需要依赖ADF调度。

实现步骤： #

1. 创建「Blob触发」类型的Azure Function，配置触发路径为你存放XLSX的Blob容器目录。
2. 在Function的requirements.txt中添加依赖：

pandas==2.1.4
azure-storage-blob==12.19.0
openpyxl==3.1.2

3. 编写Function代码：

import logging
import azure.functions as func
import pandas as pd
from azure.storage.blob import BlobServiceClient
import tempfile
import os

def main(myblob: func.InputStream):
    logging.info(f"Processing blob: {myblob.name}")

    # 从环境变量获取Blob连接字符串
    connect_str = os.environ["AzureWebJobsStorage"]
    container_name = myblob.name.split('/')[0]
    output_blob_name = myblob.name.replace(".xlsx", ".csv")

    # 初始化Blob客户端
    blob_client = BlobServiceClient.from_connection_string(connect_str).get_container_client(container_name)

    # 临时文件处理
    with tempfile.NamedTemporaryFile(suffix=".xlsx", delete=False) as temp_xlsx:
        temp_xlsx.write(myblob.read())
        temp_xlsx_path = temp_xlsx.name

    # 转换CSV
    df = pd.read_excel(temp_xlsx_path, engine="openpyxl")
    with tempfile.NamedTemporaryFile(suffix=".csv", delete=False) as temp_csv:
        df.to_csv(temp_csv, index=False, encoding="utf-8-sig")
        temp_csv_path = temp_csv.name

    # 上传回Blob
    with open(temp_csv_path, "rb") as f:
        blob_client.upload_blob(output_blob_name, f, overwrite=True)

    # 清理临时文件
    os.remove(temp_xlsx_path)
    os.remove(temp_csv_path)

优缺点： #

• ✅ 事件驱动，无需手动调度，上传即处理
• ✅ 按需付费，无运行时不产生成本
• ❌ 调试复杂度略高，需要熟悉Azure Function的部署和日志排查

选型建议 #

• 如果只是简单转格式、文件不大：优先选方案1（ADF Python活动），和现有流程整合最顺畅
• 如果有大量文件或需要复杂数据清洗：选方案2（Databricks），性能和扩展性更好
• 想要实时自动化、不需要ADF调度：选方案3（Azure Function），事件触发更灵活

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