如何通过Azure Data Factory Python SDK编程解压Blob

要实现复制Blob时自动解压，核心是在源读取配置中添加ZipDeflate解压规则，而非修改目标数据集。以下是具体修改步骤：

1. 导入必要的SDK类 #

除原有类外，新增压缩相关的配置类：

from azure.mgmt.datafactory.models import (
    BlobSource, BlobSink, DatasetReference, CopyActivity,
    PipelineResource, CompressionReadSettings, ZipDeflateReadSettings
)

2. 修改Blob源的配置 #

替换你原有的blob_source = BlobSource()代码，添加解压规则：

blob_source = BlobSource(
    compression=CompressionReadSettings(
        type="ZipDeflate",
        store_settings=ZipDeflateReadSettings(
            preserve_zip_file_name_as_folder=False  # 可选：设为True会将解压内容放在以压缩包命名的文件夹中
        )
    )
)

3. 完整修改后的复制活动代码 #

替换你原有的复制活动创建代码，管道创建、运行部分保持不变：

# here ds_name_* variables refers to Azure datasets for source and sink, respectively
act_name = 'copyBlobtoBlob'
# 配置带解压的源
blob_source = BlobSource(
    compression=CompressionReadSettings(
        type="ZipDeflate",
        store_settings=ZipDeflateReadSettings(
            preserve_zip_file_name_as_folder=False
        )
    )
)
blob_sink = BlobSink()
dsin_ref = DatasetReference(reference_name=ds_name_in, type='DatasetReference')
dsout_ref = DatasetReference(reference_name=ds_name_out, type='DatasetReference')
copy_activity = CopyActivity(name=act_name, inputs=[dsin_ref], outputs=[dsout_ref], source=blob_source, sink=blob_sink)

# make pipeline w/ copy activity
p_name = 'copyPipeline'
params_for_pipeline = {}
p_obj = PipelineResource(activities=[copy_activity], parameters=params_for_pipeline)
p = adf_client.pipelines.create_or_update(rg_name, df_name, p_name, p_obj)

# create pipeline run
run_response = adf_client.pipelines.create_run(rg_name, df_name, p_name, parameters={})

关键说明 #

• 解压逻辑在读取源Blob时触发，因此配置需加在BlobSource中，目标数据集无需额外设置
• preserve_zip_file_name_as_folder参数：设为True时，每个压缩包的内容会被放到以压缩包文件名命名的子文件夹中；设为False则直接将内容解压到目标路径根目录
• 此配置与ADF UI中设置"压缩类型为ZipDeflate"的效果完全一致

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