Szybki start: tworzenie Azure Data Factory i przepływu danych przy użyciu Python

DOTYCZY: Azure Data Factory Azure Synapse Analytics

W tym szybkim rozpoczęciu utworzysz fabrykę danych przy użyciu Pythona. Potok danych w tej fabryce kopiuje dane z jednego folderu do innego folderu w usłudze Azure Blob Storage.

Azure Data Factory to oparta na chmurze usługa integracji danych, która umożliwia tworzenie opartych na danych przepływów pracy do organizowania i automatyzowania przenoszenia danych i przekształcania danych. Za pomocą Azure Data Factory można tworzyć i planować oparte na danych przepływy pracy nazywane potokami.

Potoki mogą pobierać dane z różnych magazynów danych. Potoki danych przetwarzają lub przekształcają dane przy użyciu usług przetwarzania, takich jak Azure HDInsight Hadoop, Spark, Azure Data Lake Analytics i Azure Machine Learning. Potoki publikują dane wyjściowe w magazynach danych, takich jak Azure Synapse Analytics dla aplikacji analizy biznesowej (BI).

Wymagania wstępne

• Konto Azure z aktywną subskrypcją. Utwórz je bezpłatnie.

• Python 3.6+.

• Konto usługi Azure Storage.

• Eksplorator usługi Azure Storage (opcjonalnie).

• Aplikacja w Microsoft Entra ID. Utwórz aplikację, wykonując kroki opisane w tym linku, używając opcji uwierzytelniania 2 (klucza tajnego aplikacji) i przypisz aplikację do roli Współautor , postępując zgodnie z instrukcjami w tym samym artykule. Zanotuj następujące wartości, jak pokazano w artykule, aby użyć ich w kolejnych krokach: Identyfikator aplikacji (klienta), wartość tajemnicy klienta i identyfikator dzierżawcy.

Tworzenie i przekazywanie pliku wejściowego

1. Uruchom program Notatnik. Skopiuj poniższy tekst i zapisz go na dysku jako plik input.txt.

   John|Doe
   Jane|Doe
   

2. Użyj narzędzi, takich jak Eksplorator usługi Azure Storage, aby utworzyć kontener adfv2tutorial i folder input w kontenerze. Następnie przekaż plik input.txt do folderu input.

Instalowanie pakietu Python

1. Otwórz terminal lub wiersz polecenia z uprawnieniami administratora. 

2. Najpierw zainstaluj pakiet Python dla zasobów zarządzania Azure:

   pip install azure-mgmt-resource
   

3. Aby zainstalować pakiet Python dla usługi Data Factory, uruchom następujące polecenie:

   pip install azure-mgmt-datafactory
   

   Zestaw SDK Python dla usługi Data Factory obsługuje Python 2.7 i 3.6 lub nowszych.

4. Aby zainstalować pakiet Python na potrzeby uwierzytelniania Azure Identity, uruchom następujące polecenie:

   pip install azure-identity
   

   Uwaga

   Pakiet "azure-identity" może mieć konflikty z pakietem "azure-cli" na niektórych wspólnych zależnościach. Jeśli wystąpi jakikolwiek problem z uwierzytelnianiem, usuń pakiet "azure-cli" i jego zależności lub użyj maszyny bez zainstalowanego pakietu "azure-cli," aby rozwiązać problem. W przypadku suwerennych chmur należy używać odpowiednich stałych specyficznych dla chmury. Zapoznaj się z Łączenie się ze wszystkimi regionami za pomocą bibliotek Azure dla Pythona w środowisku multi-cloud | Microsoft Docs dla instrukcji dotyczących łączenia się za pomocą Pythona w suwerennych chmurach.

Stwórz klienta fabryki danych

1. Utwórz plik o nazwie datafactory.py. Dołącz poniższe instrukcje, aby dodać odwołania do przestrzeni nazw.

   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. Dodaj następujące funkcje, które drukują informacje.

   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. Dodaj do metody Main poniższy kod, który tworzy wystąpienie klasy DataFactoryManagementClient. Używasz tego obiektu do stworzenia fabryki danych, usługi powiązanej, zestawów danych i przepływu. Możesz także używać tego obiektu do monitorowania szczegółów uruchomienia potoku. Ustaw zmienną subscription_id na identyfikator subskrypcji Azure. Aby uzyskać listę regionów Azure, w których usługa Data Factory jest obecnie dostępna, wybierz regiony, które Cię interesują na następującej stronie, a następnie rozwiń węzeł Analytics aby zlokalizować Data Factory: Products available by region. Magazyny danych (Azure Storage, Azure SQL Database itp.) i obliczenia (HDInsight itp.) używane przez fabrykę danych mogą znajdować się w innych regionach.

   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 Sovereign 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'}
   

Tworzenie fabryki danych

Dodaj do metody Main poniższy kod, który tworzy fabrykę danych. Jeśli grupa zasobów już istnieje, dodaj komentarz do pierwszej instrukcji 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)

Tworzenie usługi połączonej

Dodaj następujący kod do metody Main, która tworzy połączoną usługę Azure Storage.

Połączone usługi tworzy się w fabryce danych w celu połączenia magazynów danych i usług obliczeniowych z fabryką danych. W tym przewodniku Szybki start wystarczy utworzyć tylko jedną połączoną usługę Azure Storage jako źródło kopiowania i miejsce docelowe o nazwie "AzureStorageLinkedService" w tym przykładzie. Zastąp <storageaccountname> i <storageaccountkey> nazwą i kluczem konta Azure Storage.

    # 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)

Tworzenie zestawów danych

W tej sekcji utworzysz zestaw danych źródła i ujścia.

Tworzenie zestawu danych dla źródłowego obiektu blob Azure

Dodaj następujący kod do metody Main, która tworzy zestaw danych obiektów blob Azure. Aby uzyskać informacje o właściwościach zestawu danych Azure Blob, zobacz łącznik Azure Blob.

Zdefiniuj zestaw danych, który reprezentuje dane źródłowe w usłudze Azure Blob Storage. Ten zestaw danych Blob odnosi się do połączonej usługi Azure Storage, którą utworzyłeś w poprzednim kroku.

    # 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)

Tworzenie zestawu danych dla docelowego zasobu Azure Blob

Dodaj następujący kod do metody Main, która tworzy zestaw danych obiektów blob Azure. Aby uzyskać informacje o właściwościach zbioru danych Azure Blob, zobacz artykuł łącznik Azure Blob.

Definiujesz zestaw danych, który reprezentuje dane źródłowe w Azure Blob. Zestaw danych dla obiektów blob odnosi się do powiązanej usługi Azure Storage, którą utworzyłeś w poprzednim kroku.

    # 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)

Utwórz potok

Dodaj do metody Main poniższy kod, który tworzy potok z działaniem kopiowania.

    # 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)

Uruchom potok

Dodaj poniższy kod do metody Main, który wyzwala uruchomienie potoku.

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

Monitoruj uruchomienie potoku

Aby monitorować uruchomienie potoku, dodaj do metody Main następujący kod:

    # 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])

Teraz dodaj poniższą instrukcję w celu wywołania metody Main po uruchomieniu programu:

# Start the main method
main()

Pełny skrypt

Oto pełny kod Python:

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

Uruchamianie kodu

Zbuduj i uruchom aplikację, a następnie zweryfikuj przebieg potoku.

Konsola wypisuje postęp tworzenia fabryki danych, połączonej usługi, zestawów danych, potoku i uruchomienia potoku. Poczekaj na wyświetlenie szczegółów uruchomienia działania kopiowania z rozmiarem odczytanych/zapisanych danych. Następnie użyj narzędzi, takich jak Azure Storage Explorer, aby sprawdzić, czy obiekty blob zostały skopiowane do "outputBlobPath" z "inputBlobPath", jak określono w zmiennych.

Oto przykładowe dane wyjściowe:

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

Czyszczenie zasobów

Aby usunąć fabrykę danych, dodaj następujący kod do programu:

adf_client.factories.delete(rg_name, df_name)

Powiązana zawartość

Przepływ pracy w tym przykładzie kopiuje dane z jednej lokalizacji do drugiej w magazynie obiektów blob Azure. Zapoznaj się z samouczkami, aby dowiedzieć się więcej o korzystaniu z usługi Data Factory w dalszych scenariuszach.