Azure Data Factory 또는 Azure Synapse Analytics 파이프라인에서 사용자 지정 활동 사용

적용 대상: Azure Data Factory Azure Synapse Analytics

Azure Data Factory 또는 Synapse 파이프라인에서 사용할 수 있는 두 가지 유형의 작업이 있습니다.

• 데이터 이동 작업은 지원되는 원본 및 싱크 데이터 저장소 간에 데이터를 이동하는 작업입니다.
• 데이터 변환 작업을 사용하여 Azure HDInsight 및 Azure Batch 같은 컴퓨팅 서비스를 사용하여 데이터를 변환합니다.

서비스에서 지원하지 않는 데이터 저장소 간에 데이터를 이동하거나, 서비스에서 지원하지 않는 방식으로 데이터를 변환/처리하려면 고유의 데이터 이동 또는 변환 논리가 포함된 사용자 지정 작업을 만들어서 파이프라인에 해당 작업을 사용하면 됩니다. 사용자 지정 작업은 가상 머신의 Azure Batch 풀에서 사용자 지정된 코드 논리를 실행합니다.

Azure Batch 서비스를 새로 사용하는 경우 다음 문서를 참조하세요.

• Azure Batch 서비스의 개요에 대한 Azure Batch 기본 사항.
• New-AzBatchAccount cmdlet을 사용하여 Azure Batch 계정(또는) Azure 포털을 만들어 Azure 포털을 사용하여 Azure Batch 계정을 만듭니다. cmdlet 사용에 대한 자세한 지침은 PowerShell을 사용하여 Azure Batch 계정 관리 문서를 참조하세요.
• New-AzBatchPool cmdlet을 사용하여 Azure Batch 풀을 만듭니다.

UI를 사용하여 파이프라인에 사용자 지정 활동 추가

파이프라인에서 사용자 지정 작업을 사용하려면 다음 단계를 완료합니다.

1. 파이프라인 활동 창에서 사용자 지정을 검색하고 사용자 지정 작업을 파이프라인 캔버스로 끌어옵니다.

2. 아직 선택하지 않은 경우 캔버스에서 새 사용자 지정 작업을 선택합니다.

3. Azure Batch 탭을 선택하여 사용자 지정 작업을 실행할 새 Azure Batch 연결된 서비스를 선택하거나 만듭니다.

   

4. 세팅 탭을 선택하고 Azure Batch 실행할 명령과 선택적 고급 세부 정보를 지정합니다.

   

연결된 서비스 Azure Batch

다음 JSON은 Azure Batch와 연동된 서비스의 예제 정의를 나타냅니다. 자세한 내용은 지원되는 컴퓨팅 환경을 참조하세요.

{
    "name": "AzureBatchLinkedService",
    "properties": {
        "type": "AzureBatch",
        "typeProperties": {
            "accountName": "batchaccount",
            "accessKey": {
                "type": "SecureString",
                "value": "access key"
            },
            "batchUri": "https://batchaccount.region.batch.azure.com",
            "poolName": "poolname",
            "linkedServiceName": {
                "referenceName": "StorageLinkedService",
                "type": "LinkedServiceReference"
            }
        }
    }
}

Azure Batch 연결 서비스에 대한 자세한 내용은 컴퓨팅 링크 서비스 문서를 참조하세요.

사용자 지정 활동

다음 JSON 코드 조각은 간단한 사용자 지정 작업이 포함된 파이프라인을 정의합니다. 활동 정의에는 Azure Batch 연결된 서비스에 대한 참조가 있습니다.

{
  "name": "MyCustomActivityPipeline",
  "properties": {
    "description": "Custom activity sample",
    "activities": [{
      "type": "Custom",
      "name": "MyCustomActivity",
      "linkedServiceName": {
        "referenceName": "AzureBatchLinkedService",
        "type": "LinkedServiceReference"
      },
      "typeProperties": {
        "command": "helloworld.exe",
        "folderPath": "customactv2/helloworld",
        "resourceLinkedService": {
          "referenceName": "StorageLinkedService",
          "type": "LinkedServiceReference"
        }
      }
    }]
  }
}

이 샘플에서 helloworld.exe resourceLinkedService에 사용되는 Azure Storage 계정의 customactv2/helloworld 폴더에 저장된 사용자 지정 애플리케이션입니다. 사용자 지정 작업은 Azure Batch 실행할 이 사용자 지정 애플리케이션을 제출합니다. 명령을 Azure Batch 풀 노드의 대상 운영 체제에서 실행할 수 있는 기본 설정 애플리케이션으로 바꿀 수 있습니다.

다음 표에는 이 작업과 관련된 속성 이름과 설명이 나와 있습니다.

* 속성 resourceLinkedService 및 folderPath를 둘 다 지정하거나 둘 다 생략해야 합니다.

사용자 지정 활동 권한

사용자 지정 작업은 Azure Batch 자동 사용자 계정을 작업 범위의 비관리자 액세스로 설정합니다(기본 자동 사용자 사양). 자동 사용자 계정의 권한 수준은 변경할 수 없습니다. 자세한 내용은 Batch에서 사용자 계정으로 태스크 실행 | 자동 사용자 계정을 참조하세요.

명령 실행

사용자 지정 작업을 사용하여 명령을 직접 실행할 수 있습니다. 다음 예제에서는 대상 Azure Batch 풀 노드에서 "echo hello world" 명령을 실행하고 출력을 stdout에 출력합니다.

{
  "name": "MyCustomActivity",
  "properties": {
    "description": "Custom activity sample",
    "activities": [{
      "type": "Custom",
      "name": "MyCustomActivity",
      "linkedServiceName": {
        "referenceName": "AzureBatchLinkedService",
        "type": "LinkedServiceReference"
      },
      "typeProperties": {
        "command": "cmd /c echo hello world"
      }
    }]
  }
}

개체 및 속성 전달

이 샘플에서는 referenceObjects 및 extendedProperties를 사용하여 개체 및 사용자 지정 속성을 서비스에서 사용자 지정 애플리케이션에 전달하는 방법을 보여 줍니다.

{
  "name": "MyCustomActivityPipeline",
  "properties": {
    "description": "Custom activity sample",
    "activities": [{
      "type": "Custom",
      "name": "MyCustomActivity",
      "linkedServiceName": {
        "referenceName": "AzureBatchLinkedService",
        "type": "LinkedServiceReference"
      },
      "typeProperties": {
        "command": "SampleApp.exe",
        "folderPath": "customactv2/SampleApp",
        "resourceLinkedService": {
          "referenceName": "StorageLinkedService",
          "type": "LinkedServiceReference"
        },
        "referenceObjects": {
          "linkedServices": [{
            "referenceName": "AzureBatchLinkedService",
            "type": "LinkedServiceReference"
          }]
        },
        "extendedProperties": {          
          "connectionString": {
            "type": "SecureString",
            "value": "aSampleSecureString"
          },
          "PropertyBagPropertyName1": "PropertyBagValue1",
          "propertyBagPropertyName2": "PropertyBagValue2",
          "dateTime1": "2015-04-12T12:13:14Z"
        }
      }
    }]
  }
}

작업이 실행될 때 referenceObjects 및 extendedProperties는 SampleApp.exe와 동일한 실행 폴더에 배포되는 다음 파일에 저장됩니다.

• activity.json

  extendedProperties는 사용자 지정 활동의 속성 정보를 저장하는 데 사용됩니다.

• linkedServices.json

  referenceObjects 속성에 정의된 연결된 서비스의 배열을 저장합니다.

• datasets.json

  referenceObjects 속성에 정의된 데이터 세트의 배열을 저장합니다.

다음 샘플 코드는 SampleApp.exe가 JSON 파일에서 필요한 정보에 액세스할 수 있는 방법을 보여줍니다.

using Newtonsoft.Json;
using System;
using System.IO;

namespace SampleApp
{
    class Program
    {
        static void Main(string[] args)
        {
            //From Extend Properties
            dynamic activity = JsonConvert.DeserializeObject(File.ReadAllText("activity.json"));
            Console.WriteLine(activity.typeProperties.extendedProperties.connectionString.value);

            // From LinkedServices
            dynamic linkedServices = JsonConvert.DeserializeObject(File.ReadAllText("linkedServices.json"));
            Console.WriteLine(linkedServices[0].properties.typeProperties.accountName);
        }
    }
}

실행 출력 검색

다음 PowerShell 명령을 사용하여 파이프라인 실행을 시작할 수 있습니다.

$runId = Invoke-AzDataFactoryV2Pipeline -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -PipelineName $pipelineName

파이프라인을 실행하는 경우 다음 명령을 사용하여 실행 출력을 확인할 수 있습니다.

while ($True) {
    $result = Get-AzDataFactoryV2ActivityRun -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -PipelineRunId $runId -RunStartedAfter (Get-Date).AddMinutes(-30) -RunStartedBefore (Get-Date).AddMinutes(30)

    if(!$result) {
        Write-Host "Waiting for pipeline to start..." -foregroundcolor "Yellow"
    }
    elseif (($result | Where-Object { $_.Status -eq "InProgress" } | Measure-Object).count -ne 0) {
        Write-Host "Pipeline run status: In Progress" -foregroundcolor "Yellow"
    }
    else {
        Write-Host "Pipeline '"$pipelineName"' run finished. Result:" -foregroundcolor "Yellow"
        $result
        break
    }
    ($result | Format-List | Out-String)
    Start-Sleep -Seconds 15
}

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

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

사용자 지정 애플리케이션의 stdout 및 stderr은 작업의 GUID를 사용하여 Azure Batch 연결된 서비스를 만들 때 정의한 Azure Storage 연결된 서비스의 adfjobs 컨테이너에 저장됩니다. 다음 코드 조각에 나와 있는 것처럼 작업 실행 출력에서 자세한 경로를 얻을 수 있습니다.

Pipeline ' MyCustomActivity' run finished. Result:

ResourceGroupName : resourcegroupname
DataFactoryName   : datafactoryname
ActivityName      : MyCustomActivity
PipelineRunId     : xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
PipelineName      : MyCustomActivity
Input             : {command}
Output            : {exitcode, outputs, effectiveIntegrationRuntime}
LinkedServiceName :
ActivityRunStart  : 10/5/2017 3:33:06 PM
ActivityRunEnd    : 10/5/2017 3:33:28 PM
DurationInMs      : 21203
Status            : Succeeded
Error             : {errorCode, message, failureType, target}

Activity Output section:
"exitcode": 0
"outputs": [
  "https://<container>.blob.core.windows.net/adfjobs/<GUID>/output/stdout.txt",
  "https://<container>.blob.core.windows.net/adfjobs/<GUID>/output/stderr.txt"
]
"effectiveIntegrationRuntime": "DefaultIntegrationRuntime (East US)"
Activity Error section:
"errorCode": ""
"message": ""
"failureType": ""
"target": "MyCustomActivity"

다운스트림 작업에서 stdout.txt의 콘텐츠를 사용하려는 경우 “@activity('MyCustomActivity').output.outputs[0]” 식에서 stdout.txt 파일의 경로를 가져올 수 있습니다.

다른 활동으로 출력 전달

사용자 지정 활동의 코드에서 사용자 지정 값을 다시 서비스에 보낼 수 있습니다. 그럼으로써 사용자 애플리케이션에서 outputs.json에 해당 값을 작성할 수 있습니다. 서비스는 outputs.json의 콘텐츠를 복사하여 customOutput 속성 값으로 작업 출력에 추가합니다. (크기는 2MB로 제한됩니다.) 다운스트림 작업에서 outputs.json의 콘텐츠를 사용하려는 경우 @activity('<MyCustomActivity>').output.customOutput 식을 사용하여 값을 가져올 수 있습니다.

SecureString 결과 가져오기

이 문서의 일부 예에 표시된 대로 SecureString 유형으로 지정된 민감한 속성 값은 사용자 인터페이스의 모니터링 탭에서 마스크 처리됩니다. 그러나 실제 파이프라인 실행에서는 SecureString 속성이 일반 텍스트로 activity.json 파일 내에서 JSON으로 serialize됩니다. 예시:

"extendedProperties": {
  "connectionString": {
    "type": "SecureString",
    "value": "aSampleSecureString"
  }
}

이 직렬화는 진정으로 안전하지 않으며 보안을 위해 설계된 것이 아닙니다. 의도는 모니터링 탭의 값을 마스킹하는 서비스에 대한 힌트입니다.

사용자 지정 활동에서 SecureString 유형의 속성에 액세스하려면 .EXE와 같은 폴더에 있는 activity.json 파일을 읽고, JSON을 역직렬화한 다음, JSON 속성(extendedProperties => [propertyName] => 값)에 액세스합니다.

Azure Batch 자동 크기 조정

또한 autoscale 기능을 사용하여 Azure Batch 풀을 만들 수도 있습니다. 예를 들어 0개의 전용 VM과 보류 중인 작업 수에 따라 자동 크기 조정 수식이 있는 Azure 일괄 처리 풀을 만들 수 있습니다.

여기에 나오는 샘플 수식은 다음과 같은 동작을 구현합니다. 풀이 처음 만들어질 때는 VM 1개로 시작합니다. $PendingTasks 메트릭은 실행되거나 큐에 대기 중인 활성 상태의 작업 수를 정의합니다. 이 수식은 지난 180초 동안에서 보류 중인 작업의 평균 수를 찾은 후 그에 따라 TargetDedicated를 설정합니다. 또한 TargetDedicated가 25개의 VM을 초과하지 않도록 합니다. 따라서 새 작업이 제출됨에 따라 풀이 자동으로 커지고, 작업이 완료되면 VM이 하나씩 자동으로 해제되어 그 VM 수가 감소합니다. startingNumberOfVMs 및 maxNumberofVMs은 요구에 맞게 조정될 수 있습니다.

자동 크기 조정 수식:

startingNumberOfVMs = 1;
maxNumberofVMs = 25;
pendingTaskSamplePercent = $PendingTasks.GetSamplePercent(180 * TimeInterval_Second);
pendingTaskSamples = pendingTaskSamplePercent < 70 ? startingNumberOfVMs : avg($PendingTasks.GetSample(180 * TimeInterval_Second));
$TargetDedicated=min(maxNumberofVMs,pendingTaskSamples);

자세한 내용은 Azure Batch 풀에서 컴퓨팅 노드 크기 조정 참조하세요.

풀에 기본 autoScaleEvaluationInterval이 사용되는 경우, Batch 서비스가 사용자 지정 작업을 실행하기 전에 VM을 준비하는 데 15~30분이 소요될 수 있습니다. 풀에 다른 autoScaleEvaluationInterval이 사용되는 경우, Batch 서비스는 autoScaleEvaluationInterval +10분이 소요될 수 있습니다.

관련 콘텐츠

다른 방법으로 데이터를 변환하는 방법을 설명하는 다음 문서를 참조하세요.

• U-SQL 작업
• Hive 작업
• Pig 작업
• MapReduce 작업
• Hadoop 스트리밍 작업
• Spark 작업
• 저장 프로시저 작업