Azure Data Factory 和 Synapse Analytics 中的 Until 活動

發行項
05/02/2024

適用於：Azure Data Factory Azure Synapse Analytics

提示

試用 Microsoft Fabric 中的 Data Factory (部分機器翻譯)，這是適用於企業的全方位分析解決方案。 Microsoft Fabric (部分機器翻譯) 涵蓋從資料移動到資料科學、即時分析、商業智慧和報告的所有項目。了解如何免費開始新的試用 (部分機器翻譯)！

Until 活動所提供的功能，與 do-until 迴圈結構在程式設計語言中提供的功能相同。它會以迴圈的方式執行一系列活動，直到與該活動相關聯的條件評估為 True 為止。如果內部活動失敗，則 Until 活動不會停止。您可以指定 Until 活動的逾時值。

使用 UI 建立 Until 活動

若要在管線中使用 Until 活動，請完成下列步驟：

在管線 [活動] 窗格中搜尋 [Until]，然後將 Until 活動拖曳至管線畫布。
在畫布上選取 Until 活動 (如未選取) 和其 [設定] 索引標籤以編輯詳細資料。
輸入將在執行 Until 活動中定義的所有子活動之後評估的運算式。如果運算式評估為 false，Until 活動會再次執行其所有子活動。當評估為 true 時，Until 活動將會完成。運算式可以是常值字串運算式，或是任何動態運算式、函式、系統變數或其他活動輸出的組合。下列範例會檢查先前定義管線陣列變數 (稱為 TestVariable) 的值，以查看其是否評估為 ['done']。
將直接選取 Until 活動上的 [編輯活動] 按鈕，或選取 [活動] 索引標籤，在該處選取活動，來定義 Until 活動將執行的活動。即會顯示新的活動編輯器窗格，您可以在其中新增可供 Until 活動執行的任何活動。在此範例中，設定變數活動只會將上述運算式中參考的變數值設定為 ['done']，因此 Until 活動的運算式會在第一次執行時為 true，而且 Until 活動將會停止。在真實世界使用中，您可以檢查所需的任何條件，而且 Until 活動會在每次評估運算式時繼續執行其子活動，直到符合條件為止。

語法

{
    "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"
}

類型屬性

屬性	說明	允許的值	必要
NAME	`Until` 活動的名稱。	String	Yes
type	必須設為 Until。	String	Yes
expression	必須評估為 true 或 false 的運算式	運算式。	Yes
timeout	do-until 迴圈在經過於此指定的時間之後便會逾時。	字串。 `d.hh:mm:ss` (或) `hh:mm:ss`。預設值是 7 天。最大值為 90 天。	No
活動	會在運算式評估為 `true` 之前持續執行的一系列活動。	一系列的活動。	Yes

範例 1

注意

本節提供 JSON 定義和 PowerShell 命令範例，以供執行管線。針對使用 Azure PowerShell 和 JSON 定義來建立管線，如需包含逐步指示的逐步解說，請參閱教學課程：使用 Azure PowerShell 來建立資料處理站。

具有 Until 活動的管線

在此範例中，管線有兩個活動：Until 與 Wait。 Wait 活動會先等候一段指定的時間，然後再以迴圈的方式執行 Web 活動。若要了解運算式和函式，請參閱運算式語言和函式。

{
    "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"
            }
        ]
    }
}

範例 2

此範例中的管線會以迴圈的方式，將資料從輸入資料夾複製到輸出資料夾。在重複參數的值設為 false，或是在逾時於一分鐘後生效時，迴圈便會終止。

具有 Until 活動的管線 (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)

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

參數化 Azure Blob 資料集 (BlobDataset.json)

管線會將 folderPath 設定為管線的 outputPath1 或 outputPath2 參數。

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

管線參數 JSON (PipelineParameters.json)

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

PowerShell 命令

注意

建議您使用 Azure Az PowerShell 模組來與 Azure 互動。請參閱安裝 Azure PowerShell 以開始使用。若要了解如何移轉至 Az PowerShell 模組，請參閱將 Azure PowerShell 從 AzureRM 移轉至 Az。

這些命令假設您已將 JSON 檔案儲存至下列資料夾：C:\ADF。

Connect-AzAccount
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 "East US" -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
}

查看其他支援的控制流程活動：