Início Rápido: Criar um data factory e pipeline usando o Python

APLICA-SE A: Azure Data Factory Azure Synapse Analytics

Dica

Experimente o Data Factory no Microsoft Fabric, uma solução de análise tudo-em-um para empresas. O Microsoft Fabric abrange desde movimentação de dados até ciência de dados, análise em tempo real, business intelligence e relatórios. Saiba como iniciar uma avaliação gratuita!

Neste início rápido, você criará um data factory usando Python. O pipeline nesse data factory copia dados de uma pasta para outra no Armazenamento de Blobs do Azure.

O Azure Data Factory é um serviço de integração de dados baseado em nuvem que permite que você crie fluxos de trabalho controlados por dados para orquestrar e automatizar a movimentação e a transformação de dados. Usando o Azure Data Factory, você pode criar e agendar fluxos de trabalho controlados por dados, chamados de pipelines.

Pipelines podem ingerir dados de armazenamentos de dados diferentes. Pipelines processam ou transformam dados usando serviços de computação, como o Azure HDInsight Hadoop, Spark, Azure Data Lake Analytics e Azure Machine Learning. Os pipelines publicam dados de saída em armazenamentos de dados como o Azure Synapse Analytics para aplicativos de BI (business intelligence).

Pré-requisitos

• Uma conta do Azure com uma assinatura ativa. Crie um gratuitamente.

• O Python 3.6+.

• Uma conta de Armazenamento do Azure.

• Gerenciador de Armazenamento do Azure (opcional).

• Um aplicativo no Microsoft Entra ID. Crie o aplicativo seguindo as etapas neste link, usando a Opção de Autenticação 2 (segredo do aplicativo) e atribua o aplicativo à função Colaborador seguindo as instruções no mesmo artigo. Anote os seguintes valores, conforme mostrado no artigo, para usá-los em etapas posteriores: ID do aplicativo (cliente), valor do segredo do cliente e ID do locatário.

Criar e carregar um arquivo de entrada

1. Inicie o Bloco de notas. Copie o texto a seguir e salve-o como um arquivo input.txt no disco.

   John|Doe
   Jane|Doe
   

2. Use ferramentas como o Gerenciador de Armazenamento do Azure para criar o contêiner adfv2tutorial e a pasta input no contêiner. Em seguida, carregue o arquivo input.txt na pasta input.

Instalar o pacote do Python

1. Abra um terminal ou prompt de comando com privilégios de administrador. 

2. Primeiro, instale o pacote do Python para recursos de gerenciamento do Azure:

   pip install azure-mgmt-resource
   

3. Para instalar o pacote do Python para o Data Factory, execute o seguinte comando:

   pip install azure-mgmt-datafactory
   

   O SDK do Python para o Data Factory é compatível com o Python 2.7 e 3.6+.

4. Para instalar o pacote do Python para a autenticação da Identidade do Azure, execute o seguinte comando:

   pip install azure-identity
   

   Observação

   O pacote "azure-identity" pode ter conflitos com "azure-cli" em algumas dependências comuns. Se você encontrar qualquer problema de autenticação, remova "azure-cli" e as dependências dela ou use um computador limpo sem instalar o pacote "azure-cli" para fazê-lo funcionar. Para nuvens soberanas, você deve usar as constantes específicas de nuvem apropriadas. Consulte Conectar-se a todas as regiões usando bibliotecas do Azure para Várias Nuvens do Python | Microsoft Docs para obter instruções para se conectar com o Python em nuvens soberanas.

Criar um cliente data factory

1. Crie um arquivo chamado datafactory.py. Adicione as instruções a seguir para adicionar referências aos namespaces.

   from azure.identity import ClientSecretCredential 
   from azure.mgmt.resource import ResourceManagementClient
   from azure.mgmt.datafactory import DataFactoryManagementClient
   from azure.mgmt.datafactory.models import *
   from datetime import datetime, timedelta
   import time
   

2. Adicione as funções a seguir, que imprimem informações.

   def print_item(group):
       """Print an Azure object instance."""
       print("\tName: {}".format(group.name))
       print("\tId: {}".format(group.id))
       if hasattr(group, 'location'):
           print("\tLocation: {}".format(group.location))
       if hasattr(group, 'tags'):
           print("\tTags: {}".format(group.tags))
       if hasattr(group, 'properties'):
           print_properties(group.properties)
   
   def print_properties(props):
       """Print a ResourceGroup properties instance."""
       if props and hasattr(props, 'provisioning_state') and props.provisioning_state:
           print("\tProperties:")
           print("\t\tProvisioning State: {}".format(props.provisioning_state))
       print("\n\n")
   
   def print_activity_run_details(activity_run):
       """Print activity run details."""
       print("\n\tActivity run details\n")
       print("\tActivity run status: {}".format(activity_run.status))
       if activity_run.status == 'Succeeded':
           print("\tNumber of bytes read: {}".format(activity_run.output['dataRead']))
           print("\tNumber of bytes written: {}".format(activity_run.output['dataWritten']))
           print("\tCopy duration: {}".format(activity_run.output['copyDuration']))
       else:
           print("\tErrors: {}".format(activity_run.error['message']))
   

3. Adicione o código a seguir, que cria uma instância da classe DataFactoryManagementClient, ao método Main. Você usa esse objeto para criar o data factory, o serviço vinculado, os conjuntos de dados e o pipeline. Você também pode usar esse objeto para monitorar os detalhes da execução de pipeline. Defina a variável subscription_id para a ID da assinatura do Azure. Para obter uma lista de regiões do Azure no qual o Data Factory está disponível no momento, selecione as regiões que relevantes para você na página a seguir e, em seguida, expanda Análise para localizar Data Factory: Produtos disponíveis por região. Os armazenamentos de dados (Armazenamento do Azure, Banco de Dados SQL do Azure, etc.) e serviços de computação (HDInsight, etc.) usados pelo data factory podem estar em outras regiões.

   def main():
   
       # Azure subscription ID
       subscription_id = '<subscription ID>'
   
       # This program creates this resource group. If it's an existing resource group, comment out the code that creates the resource group
       rg_name = '<resource group>'
   
       # The data factory name. It must be globally unique.
       df_name = '<factory name>'
   
       # Specify your Active Directory client ID, client secret, and tenant ID
       credentials = ClientSecretCredential(client_id='<Application (client) ID>', client_secret='<client secret value>', tenant_id='<tenant ID>') 
   
       # Specify following for Soverign Clouds, import right cloud constant and then use it to connect.
       # from msrestazure.azure_cloud import AZURE_PUBLIC_CLOUD as CLOUD
       # credentials = DefaultAzureCredential(authority=CLOUD.endpoints.active_directory, tenant_id=tenant_id)
   
       resource_client = ResourceManagementClient(credentials, subscription_id)
       adf_client = DataFactoryManagementClient(credentials, subscription_id)
   
       rg_params = {'location':'westus'}
       df_params = {'location':'westus'}
   

Criar uma data factory

Adicione o código a seguir, que cria um data factory, ao método Main. Se seu grupo de recursos já existir, comente a primeira instrução create_or_update.

    # create the resource group
    # comment out if the resource group already exits
    resource_client.resource_groups.create_or_update(rg_name, rg_params)

    #Create a data factory
    df_resource = Factory(location='westus')
    df = adf_client.factories.create_or_update(rg_name, df_name, df_resource)
    print_item(df)
    while df.provisioning_state != 'Succeeded':
        df = adf_client.factories.get(rg_name, df_name)
        time.sleep(1)

Criar um serviço vinculado

Adicione o código a seguir, que cria um serviço vinculado do Armazenamento do Azure, ao método Main.

Os serviços vinculados são criados em um data factory para vincular seus armazenamentos de dados e serviços de computação ao data factory. Neste guia de início rápido, você só precisa criar um serviço vinculado do Armazenamento do Azure tanto como a origem da cópia quanto como o repositório de coletor, denominado "AzureStorageLinkedService" na amostra. Substitua <storageaccountname> e <storageaccountkey> pelo nome e chave da conta de Armazenamento do Azure.

    # Create an Azure Storage linked service
    ls_name = 'storageLinkedService001'

    # IMPORTANT: specify the name and key of your Azure Storage account.
    storage_string = SecureString(value='DefaultEndpointsProtocol=https;AccountName=<account name>;AccountKey=<account key>;EndpointSuffix=<suffix>')

    ls_azure_storage = LinkedServiceResource(properties=AzureStorageLinkedService(connection_string=storage_string)) 
    ls = adf_client.linked_services.create_or_update(rg_name, df_name, ls_name, ls_azure_storage)
    print_item(ls)

Criar conjuntos de dados

Nesta seção, você criará dois conjuntos de dados: um para a origem e o outro para o coletor.

Criar um conjunto de dados para o Blob do Azure de origem

Adicione o código a seguir, que cria um Conjunto de Dados do Blob do Azure, ao método Main. Para obter informações sobre as propriedades do conjunto de dados do Blob do Azure, confira o artigo sobre o conector do Blob do Azure.

Você define um conjunto de dados que representa os dados de origem no Blob do Azure. Esse conjunto de dados de Blob refere-se ao serviço vinculado do Armazenamento do Azure que você criou na etapa anterior.

    # Create an Azure blob dataset (input)
    ds_name = 'ds_in'
    ds_ls = LinkedServiceReference(type="LinkedServiceReference",reference_name=ls_name)
    blob_path = '<container>/<folder path>'
    blob_filename = '<file name>'
    ds_azure_blob = DatasetResource(properties=AzureBlobDataset(
        linked_service_name=ds_ls, folder_path=blob_path, file_name=blob_filename)) 
    ds = adf_client.datasets.create_or_update(
        rg_name, df_name, ds_name, ds_azure_blob)
    print_item(ds)

Criar um conjunto de dados para o Blob do Azure de coletor

Adicione o código a seguir, que cria um Conjunto de Dados do Blob do Azure, ao método Main. Para obter informações sobre as propriedades do conjunto de dados do Blob do Azure, confira o artigo sobre o conector do Blob do Azure.

Você define um conjunto de dados que representa os dados de origem no Blob do Azure. Esse conjunto de dados de Blob refere-se ao serviço vinculado do Armazenamento do Azure que você criou na etapa anterior.

    # Create an Azure blob dataset (output)
    dsOut_name = 'ds_out'
    output_blobpath = '<container>/<folder path>'
    dsOut_azure_blob = DatasetResource(properties=AzureBlobDataset(linked_service_name=ds_ls, folder_path=output_blobpath))
    dsOut = adf_client.datasets.create_or_update(
        rg_name, df_name, dsOut_name, dsOut_azure_blob)
    print_item(dsOut)

Criar um pipeline

Adicione o código a seguir, que cria um pipeline com uma atividade de cópia, ao método Main.

    # Create a copy activity
    act_name = 'copyBlobtoBlob'
    blob_source = BlobSource()
    blob_sink = BlobSink()
    dsin_ref = DatasetReference(reference_name=ds_name)
    dsOut_ref = DatasetReference(reference_name=dsOut_name)
    copy_activity = CopyActivity(name=act_name,inputs=[dsin_ref], outputs=[dsOut_ref], source=blob_source, sink=blob_sink)

    #Create a pipeline with the copy activity
    
    #Note1: To pass parameters to the pipeline, add them to the json string params_for_pipeline shown below in the format { “ParameterName1” : “ParameterValue1” } for each of the parameters needed in the pipeline.
    #Note2: To pass parameters to a dataflow, create a pipeline parameter to hold the parameter name/value, and then consume the pipeline parameter in the dataflow parameter in the format @pipeline().parameters.parametername.
    
    p_name = 'copyPipeline'
    params_for_pipeline = {}

    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)
    print_item(p)

Criar uma execução de pipeline

Adicione o código a seguir, que dispara uma execução de pipeline, ao método Main.

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

Monitorar uma execução de pipeline

Para monitorar a execução de pipeline, adicione o código a seguir ao método Main:

    # Monitor the pipeline run
    time.sleep(30)
    pipeline_run = adf_client.pipeline_runs.get(
        rg_name, df_name, run_response.run_id)
    print("\n\tPipeline run status: {}".format(pipeline_run.status))
    filter_params = RunFilterParameters(
        last_updated_after=datetime.now() - timedelta(1), last_updated_before=datetime.now() + timedelta(1))
    query_response = adf_client.activity_runs.query_by_pipeline_run(
        rg_name, df_name, pipeline_run.run_id, filter_params)
    print_activity_run_details(query_response.value[0])

Agora, adicione a seguinte instrução para invocar o método principal quando o programa é executado:

# Start the main method
main()

Script completo

Aqui está o código Python completo:

from azure.identity import ClientSecretCredential 
from azure.mgmt.resource import ResourceManagementClient
from azure.mgmt.datafactory import DataFactoryManagementClient
from azure.mgmt.datafactory.models import *
from datetime import datetime, timedelta
import time

def print_item(group):
    """Print an Azure object instance."""
    print("\tName: {}".format(group.name))
    print("\tId: {}".format(group.id))
    if hasattr(group, 'location'):
        print("\tLocation: {}".format(group.location))
    if hasattr(group, 'tags'):
        print("\tTags: {}".format(group.tags))
    if hasattr(group, 'properties'):
        print_properties(group.properties)

def print_properties(props):
    """Print a ResourceGroup properties instance."""
    if props and hasattr(props, 'provisioning_state') and props.provisioning_state:
        print("\tProperties:")
        print("\t\tProvisioning State: {}".format(props.provisioning_state))
    print("\n\n")

def print_activity_run_details(activity_run):
    """Print activity run details."""
    print("\n\tActivity run details\n")
    print("\tActivity run status: {}".format(activity_run.status))
    if activity_run.status == 'Succeeded':
        print("\tNumber of bytes read: {}".format(activity_run.output['dataRead']))
        print("\tNumber of bytes written: {}".format(activity_run.output['dataWritten']))
        print("\tCopy duration: {}".format(activity_run.output['copyDuration']))
    else:
        print("\tErrors: {}".format(activity_run.error['message']))


def main():

    # Azure subscription ID
    subscription_id = '<subscription ID>'

    # This program creates this resource group. If it's an existing resource group, comment out the code that creates the resource group
    rg_name = '<resource group>'

    # The data factory name. It must be globally unique.
    df_name = '<factory name>'

    # Specify your Active Directory client ID, client secret, and tenant ID
    credentials = ClientSecretCredential(client_id='<service principal ID>', client_secret='<service principal key>', tenant_id='<tenant ID>') 
    resource_client = ResourceManagementClient(credentials, subscription_id)
    adf_client = DataFactoryManagementClient(credentials, subscription_id)

    rg_params = {'location':'westus'}
    df_params = {'location':'westus'}
 
    # create the resource group
    # comment out if the resource group already exits
    resource_client.resource_groups.create_or_update(rg_name, rg_params)

    # Create a data factory
    df_resource = Factory(location='westus')
    df = adf_client.factories.create_or_update(rg_name, df_name, df_resource)
    print_item(df)
    while df.provisioning_state != 'Succeeded':
        df = adf_client.factories.get(rg_name, df_name)
        time.sleep(1)

    # Create an Azure Storage linked service
    ls_name = 'storageLinkedService001'

    # IMPORTANT: specify the name and key of your Azure Storage account.
    storage_string = SecureString(value='DefaultEndpointsProtocol=https;AccountName=<account name>;AccountKey=<account key>;EndpointSuffix=<suffix>')

    ls_azure_storage = LinkedServiceResource(properties=AzureStorageLinkedService(connection_string=storage_string)) 
    ls = adf_client.linked_services.create_or_update(rg_name, df_name, ls_name, ls_azure_storage)
    print_item(ls)

    # Create an Azure blob dataset (input)
    ds_name = 'ds_in'
    ds_ls = LinkedServiceReference(type="LinkedServiceReference",reference_name=ls_name)
    blob_path = '<container>/<folder path>'
    blob_filename = '<file name>'
    ds_azure_blob = DatasetResource(properties=AzureBlobDataset(
        linked_service_name=ds_ls, folder_path=blob_path, file_name=blob_filename))
    ds = adf_client.datasets.create_or_update(
        rg_name, df_name, ds_name, ds_azure_blob)
    print_item(ds)

    # Create an Azure blob dataset (output)
    dsOut_name = 'ds_out'
    output_blobpath = '<container>/<folder path>'
    dsOut_azure_blob = DatasetResource(properties=AzureBlobDataset(linked_service_name=ds_ls, folder_path=output_blobpath))
    dsOut = adf_client.datasets.create_or_update(
        rg_name, df_name, dsOut_name, dsOut_azure_blob)
    print_item(dsOut)

    # Create a copy activity
    act_name = 'copyBlobtoBlob'
    blob_source = BlobSource()
    blob_sink = BlobSink()
    dsin_ref = DatasetReference(reference_name=ds_name)
    dsOut_ref = DatasetReference(reference_name=dsOut_name)
    copy_activity = CopyActivity(name=act_name, inputs=[dsin_ref], outputs=[
                                 dsOut_ref], source=blob_source, sink=blob_sink)

    # Create a pipeline with the 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)
    print_item(p)

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

    # Monitor the pipeline run
    time.sleep(30)
    pipeline_run = adf_client.pipeline_runs.get(
        rg_name, df_name, run_response.run_id)
    print("\n\tPipeline run status: {}".format(pipeline_run.status))
    filter_params = RunFilterParameters(
        last_updated_after=datetime.now() - timedelta(1), last_updated_before=datetime.now() + timedelta(1))
    query_response = adf_client.activity_runs.query_by_pipeline_run(
        rg_name, df_name, pipeline_run.run_id, filter_params)
    print_activity_run_details(query_response.value[0])


# Start the main method
main()

Executar o código

Compile e inicie o aplicativo, então verifique a execução do pipeline.

O console imprime o progresso de criação do data factory, do serviço vinculado, dos conjuntos de dados, do pipeline e da execução de pipeline. Aguarde até ver os detalhes de execução da atividade de cópia com o tamanho dos dados lidos/gravados. Em seguida, use ferramentas como o Gerenciador de Armazenamento do Azure para verificar se os blobs são copiados de "inputBlobPath" para "outputBlobPath" conforme você especificou nas variáveis.

Veja o exemplo de saída:

Name: <data factory name>
Id: /subscriptions/<subscription ID>/resourceGroups/<resource group name>/providers/Microsoft.DataFactory/factories/<data factory name>
Location: eastus
Tags: {}

Name: storageLinkedService
Id: /subscriptions/<subscription ID>/resourceGroups/<resource group name>/providers/Microsoft.DataFactory/factories/<data factory name>/linkedservices/storageLinkedService

Name: ds_in
Id: /subscriptions/<subscription ID>/resourceGroups/<resource group name>/providers/Microsoft.DataFactory/factories/<data factory name>/datasets/ds_in

Name: ds_out
Id: /subscriptions/<subscription ID>/resourceGroups/<resource group name>/providers/Microsoft.DataFactory/factories/<data factory name>/datasets/ds_out

Name: copyPipeline
Id: /subscriptions/<subscription ID>/resourceGroups/<resource group name>/providers/Microsoft.DataFactory/factories/<data factory name>/pipelines/copyPipeline

Pipeline run status: Succeeded
Datetime with no tzinfo will be considered UTC.
Datetime with no tzinfo will be considered UTC.

Activity run details

Activity run status: Succeeded
Number of bytes read: 18
Number of bytes written: 18
Copy duration: 4

Limpar os recursos

Para excluir o data factory, adicione o código a seguir ao programa:

adf_client.factories.delete(rg_name, df_name)

Conteúdo relacionado

O pipeline nessa amostra copia dados de uma localização para outra em um Armazenamento de Blobs do Azure. Percorra os tutoriais para saber mais sobre o uso do Data Factory em mais cenários.