在Databricks中合并Blob容器内同Schema多Part文件为单Tab文件

下面提供几种实用的实现方式，根据你的技术栈选择即可：

方法一：用Azure Data Factory（ADF）可视化实现 #

适合不需要写代码的场景，操作简单易维护：

• 先创建Blob存储链接服务，关联你的目标存储账户，源和目标可共用同一个链接服务
• 配置源数据集：选择Blob存储，路径指定到Staging文件夹，文件名用通配符*.part，根据Part文件的实际格式（比如CSV、Parquet）设置对应格式属性
• 配置目标数据集：同样选Blob存储，格式设置为分隔符文本，分隔符选制表符（\t），指定输出的单个文件名（比如merged_sales_data.tab）
• 添加复制活动：源选刚才的源数据集，目标选目标数据集，因为所有Part文件Schema一致，ADF会自动匹配字段，直接运行管道就能完成合并和格式转换

方法二：PowerShell脚本批量处理 #

适合熟悉PowerShell的管理员，灵活可控：

# 连接Azure账户
Connect-AzAccount

# 替换为你的存储账户、资源组、容器信息
$storageAccountName = "your-storage-account"
$resourceGroupName = "your-resource-group"
$containerName = "your-container"
$stagingFolderPrefix = "Staging/"
$targetBlobName = "merged_output.tab"

# 获取存储账户上下文
$storageAccountKey = (Get-AzStorageAccountKey -ResourceGroupName $resourceGroupName -Name $storageAccountName).Value[0]
$ctx = New-AzStorageContext -StorageAccountName $storageAccountName -StorageAccountKey $storageAccountKey

$headerWritten = $false
$tempTabFile = "./temp_merged.tab"

# 遍历Staging下所有Part文件
Get-AzStorageBlob -Container $containerName -Prefix $stagingFolderPrefix -Context $ctx | Where-Object { $_.Name -like "*.part" } | ForEach-Object {
    # 临时下载当前Part文件
    Get-AzStorageBlobContent -Blob $_.Name -Container $containerName -Context $ctx -Destination "./temp.part" -Force | Out-Null
    
    $contentLines = Get-Content "./temp.part" -Encoding UTF8
    if (-not $headerWritten) {
        # 首次写入表头
        $contentLines[0] | Out-File -FilePath $tempTabFile -Encoding UTF8
        $headerWritten = $true
        # 写入当前文件的内容行（跳过表头）
        $contentLines[1..($contentLines.Count-1)] | Out-File -FilePath $tempTabFile -Encoding UTF8 -Append
    } else {
        # 跳过表头，直接写入内容行
        $contentLines[1..($contentLines.Count-1)] | Out-File -FilePath $tempTabFile -Encoding UTF8 -Append
    }

    # 上传更新后的合并文件到Blob
    Set-AzStorageBlobContent -File $tempTabFile -Container $containerName -Blob $targetBlobName -Context $ctx -Force | Out-Null
}

# 清理本地临时文件
Remove-Item "./temp.part" -Force
Remove-Item $tempTabFile -Force

方法三：Python脚本实现 #

适合开发人员，可自定义扩展逻辑：
先安装依赖包：pip install azure-storage-blob

from azure.storage.blob import BlobServiceClient
import os

# 替换为你的配置信息
STORAGE_CONNECTION_STRING = "your-storage-connection-string"
CONTAINER_NAME = "your-container"
STAGING_FOLDER_PREFIX = "Staging/"
TARGET_BLOB_NAME = "merged_output.tab"

# 初始化Blob客户端
blob_service_client = BlobServiceClient.from_connection_string(STORAGE_CONNECTION_STRING)
container_client = blob_service_client.get_container_client(CONTAINER_NAME)

header_written = False
temp_tab_path = "./temp_merged.tab"

# 遍历Staging目录下的所有Part文件
for blob in container_client.list_blobs(name_starts_with=STAGING_FOLDER_PREFIX):
    if blob.name.endswith(".part"):
        blob_client = container_client.get_blob_client(blob)
        # 临时下载Part文件
        with open("./temp.part", "wb") as f:
            download_stream = blob_client.download_blob()
            download_stream.readinto(f)
        
        # 读取并写入合并文件
        with open("./temp.part", "r", encoding="utf-8") as part_file, open(temp_tab_path, "a", encoding="utf-8") as tab_file:
            lines = part_file.readlines()
            if not header_written:
                tab_file.write(lines[0])
                header_written = True
                tab_file.writelines(lines[1:])
            else:
                tab_file.writelines(lines[1:])

# 上传合并后的Tab文件到Blob容器
with open(temp_tab_path, "rb") as data:
    container_client.upload_blob(name=TARGET_BLOB_NAME, data=data, overwrite=True)

# 清理本地临时文件
os.remove("./temp.part")
os.remove(temp_tab_path)

注意事项 #

• 确保所有Part文件的编码一致（建议用UTF-8），避免合并后出现乱码
• 如果Part文件体积较大，优先选择ADF或分块读取的方式，防止内存溢出
• 测试阶段先选取少量文件验证Schema匹配和格式转换是否正确

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