求助：Azure Data Factory自定义活动转DBF为CSV后文件未存至存储账户

你的代码执行成功但找不到CSV文件，核心原因是转换后的CSV仅保存在Batch计算节点的本地磁盘，没有上传回Azure存储账户。以下是修改后的完整代码，包含DBF下载、转换、CSV上传到Blob的完整流程：

import csv
import os
from dbfread import DBF, FieldParser
from azure.storage.blob import BlobServiceClient

# 替换为你的Azure存储账户信息
account_name = "my_account_name"
account_key = "my_account_key"
source_container = "dbf"  # 存放DBF文件的容器
target_container = "csv"  # 存放生成CSV的容器（可与源容器相同）
dbf_blob_name = "test.dbf"
csv_blob_name = "converted_test.csv"  # 上传到Blob的CSV文件名

# 初始化Blob服务客户端
connection_string = f"DefaultEndpointsProtocol=https;AccountName={account_name};AccountKey={account_key};EndpointSuffix=core.windows.net"
blob_service_client = BlobServiceClient.from_connection_string(connection_string)

# 1. 从Blob下载DBF文件到本地临时路径
local_dbf_path = "temp.dbf"
try:
    blob_client = blob_service_client.get_blob_client(container=source_container, blob=dbf_blob_name)
    with open(local_dbf_path, "wb") as f:
        f.write(blob_client.download_blob().readall())
except Exception as e:
    print(f"下载DBF文件失败: {e}")
    exit(1)

# 2. DBF转CSV的自定义字段解析器
class CustomFieldParser(FieldParser):
    def _parse_memo_index(self, data):
        try:
            return int(data)
        except ValueError:
            return data.decode('utf-8', errors='replace')

# 3. 转换DBF到本地CSV
local_csv_path = "temp.csv"
try:
    dbf_file = DBF(local_dbf_path, parserclass=CustomFieldParser)
    with open(local_csv_path, "w", newline="", encoding="utf-8") as csvfile:
        writer = csv.writer(csvfile)
        writer.writerow(dbf_file.field_names)
        for record in dbf_file:
            writer.writerow(list(record.values()))
    print("DBF转CSV完成")
except Exception as e:
    print(f"转换或写入CSV失败: {e}")
    exit(1)

# 4. 将本地CSV上传到Blob存储
try:
    blob_client = blob_service_client.get_blob_client(container=target_container, blob=csv_blob_name)
    with open(local_csv_path, "rb") as f:
        blob_client.upload_blob(f, overwrite=True)
    print(f"CSV文件已上传到Blob: {target_container}/{csv_blob_name}")
except Exception as e:
    print(f"上传CSV到Blob失败: {e}")
    exit(1)

# 清理本地临时文件（可选，避免占用Batch节点磁盘空间）
try:
    os.remove(local_dbf_path)
    os.remove(local_csv_path)
except Exception as e:
    print(f"清理临时文件失败: {e}")

关键说明 #

• 新增上传逻辑：代码最后添加了将本地生成的CSV上传到目标Blob容器的步骤，这是原代码缺失的核心部分
• 临时文件清理：可选步骤，避免Batch计算节点磁盘被临时文件占用
• 依赖包安装：确保Batch计算节点已安装所需依赖，可通过Batch任务的command line执行安装命令：

  pip install dbfread azure-storage-blob
  

ADF自定义活动配置要点 #

• 在ADF自定义活动中，指定Batch账户和池时，确保任务的命令行包含依赖安装步骤（如果节点未预装）
• 确认存储账户的访问密钥或SAS令牌权限足够（需包含Blob的读、写权限）

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