Azure 数据工厂中的 Until 活动Until activity in Azure Data Factory

适用于: Azure 数据工厂 Azure Synapse Analytics

Until 活动提供的功能与 do-until 循环结构以编程语言提供的功能相同。The Until activity provides the same functionality that a do-until looping structure provides in programming languages. 它在循环中将执行一组活动,直到与活动相关联的条件的计算结果为 true。It executes a set of activities in a loop until the condition associated with the activity evaluates to true. 你可以在数据工厂中为 Until 活动指定超时值。You can specify a timeout value for the until activity in Data Factory.

语法Syntax

{
    "type": "Until",
    "typeProperties": {
        "expression":  {
            "value":  "<expression that evaluates to true or false>", 
            "type": "Expression"
        },
        "timeout": "<time out for the loop. for example: 00:01:00 (1 minute)>",
        "activities": [
            {
                "<Activity 1 definition>"
            },
            {
                "<Activity 2 definition>"
            },
            {
                "<Activity N definition>"
            }
        ]
    },
    "name": "MyUntilActivity"
}

Type 属性Type properties

propertiesProperty 说明Description 允许的值Allowed values 必选Required
namename Until 活动的名称。Name of the Until activity. StringString Yes
typetype 必须设置为 Until 。Must be set to Until. StringString Yes
表达式expression 计算结果必须为 true 或 false 的表达式Expression that must evaluate to true or false 表达式。Expression. Yes
timeouttimeout 此处在指定的时间之后 do-until 循环超时。The do-until loop times out after the specified time here. 字符串。String. d.hh:mm:sshh:mm:ssd.hh:mm:ss (or) hh:mm:ss. 默认值为 7 天。The default value is 7 days. 最大值为 90 天。Maximum value is: 90 days. No
活动Activities 在表达式计算结果为 true 前将执行的活动集。Set of activities that are executed until expression evaluates to true. 活动数组Array of activities. Yes

示例 1Example 1

备注

本部分提供运行管道的 JSON 定义和示例 PowerShell 命令。This section provides JSON definitions and sample PowerShell commands to run the pipeline. 有关使用 Azure PowerShell 和 JSON 定义创建数据工厂管道的分步说明演练,请参阅教程:使用 Azure PowerShell 创建数据工厂For a walkthrough with step-by-step instructions to create a Data Factory pipeline by using Azure PowerShell and JSON definitions, see tutorial: create a data factory by using Azure PowerShell.

包含 Until 活动的管道Pipeline with Until activity

在此示例中,管道具有两个活动:UntilWaitIn this example, the pipeline has two activities: Until and Wait. 在循环中运行 Web 活动之前,Wait 活动会等待指定的期间。The Wait activity waits for the specified period of time before running the Web activity in the loop. 若要了解数据工厂中的表达式和函数,请参阅表达式语言和函数To learn about expressions and functions in Data Factory, see Expression language and functions.

{
    "name": "DoUntilPipeline",
    "properties": {
        "activities": [
            {
                "type": "Until",
                "typeProperties": {
                    "expression": {
                        "value": "@equals('Failed', coalesce(body('MyUnauthenticatedActivity')?.status, actions('MyUnauthenticatedActivity')?.status, 'null'))",
                        "type": "Expression"
                    },
                    "timeout": "00:00:01",
                    "activities": [
                        {
                            "name": "MyUnauthenticatedActivity",
                            "type": "WebActivity",
                            "typeProperties": {
                                "method": "get",
                                "url": "https://www.fake.com/",
                                "headers": {
                                    "Content-Type": "application/json"
                                }
                            },
                            "dependsOn": [
                                {
                                    "activity": "MyWaitActivity",
                                    "dependencyConditions": [ "Succeeded" ]
                                }
                            ]
                        },
                        {
                            "type": "Wait",
                            "typeProperties": {
                                "waitTimeInSeconds": 1
                            },
                            "name": "MyWaitActivity"
                        }
                    ]
                },
                "name": "MyUntilActivity"
            }
        ]
    }
}

示例 2Example 2

此示例中的管道可在循环中将数据从输入文件夹复制到输出文件夹。The pipeline in this sample copies data from an input folder to an output folder in a loop. 循环将在重复参数的值设置为 false 时终止,否则会在一分钟后超时。The loop terminates when the value for the repeat parameter is set to false or it times out after one minute.

包含 Until 活动的管道 (Adfv2QuickStartPipeline.json)Pipeline with Until activity (Adfv2QuickStartPipeline.json)

{
    "name": "Adfv2QuickStartPipeline",
    "properties": {
        "activities": [
            {
                "type": "Until",
                "typeProperties": {
                    "expression":  {
                        "value":  "@equals('false', pipeline().parameters.repeat)", 
                        "type": "Expression"
                    },
                    "timeout": "00:01:00",
                    "activities": [
                        {
                            "name": "CopyFromBlobToBlob",
                            "type": "Copy",
                            "inputs": [
                                {
                                    "referenceName": "BlobDataset",
                                    "parameters": {
                                        "path": "@pipeline().parameters.inputPath"
                                    },
                                    "type": "DatasetReference"
                                }
                            ],
                            "outputs": [
                                {
                                    "referenceName": "BlobDataset",
                                    "parameters": {
                                        "path": "@pipeline().parameters.outputPath"
                                    },
                                    "type": "DatasetReference"
                                }
                            ],
                            "typeProperties": {
                                "source": {
                                    "type": "BlobSource"
                                },
                                "sink": {
                                    "type": "BlobSink"
                                }
                            },
                            "policy": {
                                "retry": 1,
                                "timeout": "00:10:00",
                                "retryIntervalInSeconds": 60
                            }
                        }
                    ]
                },
                "name": "MyUntilActivity"
            }
        ],
        "parameters": {
            "inputPath": {
                "type": "String"
            },
            "outputPath": {
                "type": "String"
            },
            "repeat": {
                "type": "String"
            }                        
        }        
    }
}

Azure 存储链接服务 (AzureStorageLinkedService.json)Azure Storage linked service (AzureStorageLinkedService.json)

{
    "name": "AzureStorageLinkedService",
    "properties": {
        "type": "AzureStorage",
        "typeProperties": {
            "connectionString": "DefaultEndpointsProtocol=https;AccountName=<Azure Storage account name>;AccountKey=<Azure Storage account key>;EndpointSuffix=core.chinacloudapi.cn"
        }
    }
}

参数化的 Azure Blob 数据集 (BlobDataset.json)Parameterized Azure Blob dataset (BlobDataset.json)

管道将 folderPath 设置为管道参数 outputPath1 或 outputPath2 的值。The pipeline sets the folderPath to the value of either outputPath1 or outputPath2 parameter of the pipeline.

{
    "name": "BlobDataset",
    "properties": {
        "type": "AzureBlob",
        "typeProperties": {
            "folderPath": {
                "value": "@{dataset().path}",
                "type": "Expression"
            }
        },
        "linkedServiceName": {
            "referenceName": "AzureStorageLinkedService",
            "type": "LinkedServiceReference"
        },
        "parameters": {
            "path": {
                "type": "String"
            }
        }
    }
}

管道参数 JSON (PipelineParameters.json)Pipeline parameter JSON (PipelineParameters.json)

{
    "inputPath": "adftutorial/input",
    "outputPath": "adftutorial/outputUntil",
    "repeat": "true"
}

PowerShell 命令PowerShell commands

备注

本文已经过更新,以便使用 Azure Az PowerShell 模块。This article has been updated to use the Azure Az PowerShell module. 若要与 Azure 交互,建议使用的 PowerShell 模块是 Az PowerShell 模块。The Az PowerShell module is the recommended PowerShell module for interacting with Azure. 若要开始使用 Az PowerShell 模块,请参阅安装 Azure PowerShellTo get started with the Az PowerShell module, see Install Azure PowerShell. 若要了解如何迁移到 Az PowerShell 模块,请参阅 将 Azure PowerShell 从 AzureRM 迁移到 AzTo learn how to migrate to the Az PowerShell module, see Migrate Azure PowerShell from AzureRM to Az.

这些命令假设你已将 JSON 文件保存到文件夹 C:\ADF。These commands assume that you have saved the JSON files into the folder: C:\ADF.

Connect-AzAccount -Environment AzureChinaCloud
Select-AzSubscription "<Your subscription name>"

$resourceGroupName = "<Resource Group Name>"
$dataFactoryName = "<Data Factory Name. Must be globally unique>";
Remove-AzDataFactoryV2 $dataFactoryName -ResourceGroupName $resourceGroupName -force


Set-AzDataFactoryV2 -ResourceGroupName $resourceGroupName -Location "China East 2" -Name $dataFactoryName
Set-AzDataFactoryV2LinkedService -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -Name "AzureStorageLinkedService" -DefinitionFile "C:\ADF\AzureStorageLinkedService.json"
Set-AzDataFactoryV2Dataset -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -Name "BlobDataset" -DefinitionFile "C:\ADF\BlobDataset.json"
Set-AzDataFactoryV2Pipeline -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -Name "Adfv2QuickStartPipeline" -DefinitionFile "C:\ADF\Adfv2QuickStartPipeline.json"
$runId = Invoke-AzDataFactoryV2Pipeline -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -PipelineName "Adfv2QuickStartPipeline" -ParameterFile C:\ADF\PipelineParameters.json

while ($True) {
    $run = Get-AzDataFactoryV2PipelineRun -ResourceGroupName $resourceGroupName -DataFactoryName $DataFactoryName -PipelineRunId $runId

    if ($run) {
        if ($run.Status -ne 'InProgress') {
            Write-Host "Pipeline run finished. The status is: " $run.Status -foregroundcolor "Yellow"
            $run
            break
        }
        Write-Host  "Pipeline is running...status: InProgress" -foregroundcolor "Yellow"
        Write-Host "Activity run details:" -foregroundcolor "Yellow"
        $result = Get-AzDataFactoryV2ActivityRun -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -PipelineRunId $runId -RunStartedAfter (Get-Date).AddMinutes(-30) -RunStartedBefore (Get-Date).AddMinutes(30)
        $result

        Write-Host "Activity 'Output' section:" -foregroundcolor "Yellow"
        $result.Output -join "`r`n"
    }

    Start-Sleep -Seconds 15
}

后续步骤Next steps

查看数据工厂支持的其他控制流活动:See other control flow activities supported by Data Factory: