python - 如何在 Python 中为 Azure 数据工厂的 Pipeline 活动创建依赖关系

问题描述

在 Azure DataFactory 管道中，我尝试让两个 CopyActivities 顺序运行，即第一个将数据从 blob 复制到 SQL 表，然后第二个将 SQL 表复制到另一个数据库。

我尝试了下面的代码，但结果管道不依赖于活动（从工作流程图和 JSON 在 Azure UI 中检查）。当我运行管道时，我收到如下错误消息： “ErrorResponseException：模板验证失败：'1'行和'22521'列的模板操作'我的第二个活动nameScope'的'runAfter'属性包含不存在行动。巴拉巴布拉……”

在 Azure UI 中手动添加依赖项后，我可以成功运行管道。

如果有人能指出示例代码（Python/C#/Powershell）或文档，我将不胜感激。我的 Python 代码：

    def createDataFactoryRectStage(self,  
                               aPipelineName, aActivityStageName, aActivityAcquireName,
                               aRectFileName, aRectDSName,
                               aStageTableName, aStageDSName,
                               aAcquireTableName, aAcquireDSName):
    adf_client = self.__getAdfClient()

    ds_blob = AzureBlobDataset(linked_service_name = LinkedServiceReference(AZURE_DATAFACTORY_LS_BLOB_RECT), 
                               folder_path=PRJ_AZURE_BLOB_PATH_RECT, 
                               file_name = aRectFileName,
                               format = {"type": "TextFormat",
                                         "columnDelimiter": ",",
                                         "rowDelimiter": "",
                                         "nullValue": "\\N",
                                         "treatEmptyAsNull": "true",
                                         "firstRowAsHeader": "true",
                                         "quoteChar": "\"",})    
    adf_client.datasets.create_or_update(AZURE_RESOURCE_GROUP, AZURE_DATAFACTORY, aRectDSName, ds_blob)

    ds_stage= AzureSqlTableDataset(linked_service_name = LinkedServiceReference(AZURE_DATAFACTORY_LS_SQLDB_STAGE), 
                                   table_name='[dbo].[' + aStageTableName + ']')      
    adf_client.datasets.create_or_update(AZURE_RESOURCE_GROUP, AZURE_DATAFACTORY, aStageDSName, ds_stage)

    ca_blob_to_stage = CopyActivity(aActivityStageName, 
                                    inputs=[DatasetReference(aRectDSName)], 
                                    outputs=[DatasetReference(aStageDSName)], 
                                    source= BlobSource(), 
                                    sink= SqlSink(write_batch_size = AZURE_SQL_WRITE_BATCH_SIZE))

    ds_acquire= AzureSqlTableDataset(linked_service_name = LinkedServiceReference(AZURE_DATAFACTORY_LS_SQLDB_ACQUIRE), 
                                     table_name='[dbo].[' + aAcquireTableName + ']')      
    adf_client.datasets.create_or_update(AZURE_RESOURCE_GROUP, AZURE_DATAFACTORY, aAcquireDSName, ds_acquire)
    dep = ActivityDependency(ca_blob_to_stage, dependency_conditions =[DependencyCondition('Succeeded')])

    ca_stage_to_acquire = CopyActivity(aActivityAcquireName, 
                                       inputs=[DatasetReference(aStageDSName)], 
                                       outputs=[DatasetReference(aAcquireDSName)], 
                                       source= SqlSource(), 
                                       sink= SqlSink(write_batch_size = AZURE_SQL_WRITE_BATCH_SIZE),
                                       depends_on=[dep])

    p_obj = PipelineResource(activities=[ca_blob_to_stage, ca_stage_to_acquire], parameters={})

    return adf_client.pipelines.create_or_update(AZURE_RESOURCE_GROUP, AZURE_DATAFACTORY, aPipelineName, p_obj)

标签： pythondependenciesazure-data-factory

python - 如何在 Python 中为 Azure 数据工厂的 Pipeline 活动创建依赖关系

问题描述

解决方案

推荐阅读