Partager via

Gérer Azure Data Lake Analytics à l’aide de Python

Important

Azure Data Lake Analytics a pris sa retraite le 29 février 2024. Découvrez-en plus avec cette annonce.

Pour l'analyse des données, votre organisation peut utiliser Azure Synapse Analytics ou Microsoft Fabric.

Cet article explique comment gérer des comptes Azure Data Lake Analytics, des sources de données, des utilisateurs et des travaux à l’aide de Python.

Versions de Python prises en charge

  • Utilisez une version 64 bits de Python.
  • Vous pouvez utiliser la distribution Python standard trouvée au Python.org téléchargements.
  • De nombreux développeurs trouvent qu’il est pratique d’utiliser la distribution Anaconda Python.
  • Cet article a été écrit à l’aide de Python version 3.6 à partir de la distribution Python standard

Installer le Kit de développement logiciel (SDK) Azure Python

Installez les modules suivants :

  • Le module azure-mgmt-resource inclut d’autres modules Azure pour Active Directory, etc.
  • Le module azure-datalake-store inclut les opérations de système de fichiers Azure Data Lake Store.
  • Le module azure-mgmt-datalake-store inclut les opérations de gestion des comptes Azure Data Lake Store.
  • Le module azure-mgmt-datalake-analytics inclut les opérations Azure Data Lake Analytics.

Tout d’abord, vérifiez que vous disposez de la dernière pip version en exécutant la commande suivante :

python -m pip install --upgrade pip

Ce document a été écrit à l’aide pip version 9.0.1de .

Utilisez les commandes suivantes pip pour installer les modules à partir de la ligne de commande :

pip install azure-identity
pip install azure-mgmt-resource
pip install azure-datalake-store
pip install azure-mgmt-datalake-store
pip install azure-mgmt-datalake-analytics

Créer un script Python

Collez le code suivant dans le script :

# Use this only for Azure AD service-to-service authentication
#from azure.common.credentials import ServicePrincipalCredentials

# Use this only for Azure AD end-user authentication
#from azure.common.credentials import UserPassCredentials

# Required for Azure Identity
from azure.identity import DefaultAzureCredential

# Required for Azure Resource Manager
from azure.mgmt.resource.resources import ResourceManagementClient
from azure.mgmt.resource.resources.models import ResourceGroup

# Required for Azure Data Lake Store account management
from azure.mgmt.datalake.store import DataLakeStoreAccountManagementClient
from azure.mgmt.datalake.store.models import DataLakeStoreAccount

# Required for Azure Data Lake Store filesystem management
from azure.datalake.store import core, lib, multithread

# Required for Azure Data Lake Analytics account management
from azure.mgmt.datalake.analytics.account import DataLakeAnalyticsAccountManagementClient
from azure.mgmt.datalake.analytics.account.models import DataLakeAnalyticsAccount, DataLakeStoreAccountInformation

# Required for Azure Data Lake Analytics job management
from azure.mgmt.datalake.analytics.job import DataLakeAnalyticsJobManagementClient
from azure.mgmt.datalake.analytics.job.models import JobInformation, JobState, USqlJobProperties

# Required for Azure Data Lake Analytics catalog management
from azure.mgmt.datalake.analytics.catalog import DataLakeAnalyticsCatalogManagementClient

# Required for Azure Data Lake Analytics Model
from azure.mgmt.datalake.analytics.account.models import CreateOrUpdateComputePolicyParameters

# Use these as needed for your application
import logging
import getpass
import pprint
import uuid
import time

Exécutez ce script pour vérifier que les modules peuvent être importés.

Authentification

Authentification utilisateur interactive avec une fenêtre contextuelle

Cette méthode n’est pas prise en charge.

Authentification utilisateur interactive avec un code d’appareil

user = input(
    'Enter the user to authenticate with that has permission to subscription: ')
password = getpass.getpass()
credentials = UserPassCredentials(user, password)

Authentification non interactive avec SPI et secret

# Acquire a credential object for the app identity. When running in the cloud,
# DefaultAzureCredential uses the app's managed identity (MSI) or user-assigned service principal.
# When run locally, DefaultAzureCredential relies on environment variables named
# AZURE_CLIENT_ID, AZURE_CLIENT_SECRET, and AZURE_TENANT_ID.

credentials = DefaultAzureCredential()

Authentification non réactive avec l’API et un certificat

Cette méthode n’est pas prise en charge.

Variables de script courantes

Ces variables sont utilisées dans les exemples.

subid = '<Azure Subscription ID>'
rg = '<Azure Resource Group Name>'
location = '<Location>'  # i.e. 'eastus2'
adls = '<Azure Data Lake Store Account Name>'
adla = '<Azure Data Lake Analytics Account Name>'

Créer les clients

resourceClient = ResourceManagementClient(credentials, subid)
adlaAcctClient = DataLakeAnalyticsAccountManagementClient(credentials, subid)
adlaJobClient = DataLakeAnalyticsJobManagementClient(
    credentials, 'azuredatalakeanalytics.net')

Créer un groupe de ressources Azure

armGroupResult = resourceClient.resource_groups.create_or_update(
    rg, ResourceGroup(location=location))

Créer un compte Data Lake Analytics

Commencez par créer un compte de magasin.

adlsAcctResult = adlsAcctClient.account.begin_create(
	rg,
	adls,
	DataLakeStoreAccount(
		location=location)
	)
).wait()

Créez ensuite un compte ADLA qui utilise ce magasin.

adlaAcctResult = adlaAcctClient.account.create(
    rg,
    adla,
    DataLakeAnalyticsAccount(
        location=location,
        default_data_lake_store_account=adls,
        data_lake_store_accounts=[DataLakeStoreAccountInformation(name=adls)]
    )
).wait()

Soumettre un job

script = """
@a  = 
    SELECT * FROM 
        (VALUES
            ("Contoso", 1500.0),
            ("Woodgrove", 2700.0)
        ) AS 
              D( customer, amount );
OUTPUT @a
    TO "/data.csv"
    USING Outputters.Csv();
"""

jobId = str(uuid.uuid4())
jobResult = adlaJobClient.job.create(
    adla,
    jobId,
    JobInformation(
        name='Sample Job',
        type='USql',
        properties=USqlJobProperties(script=script)
    )
)

Attendre la fin d’un travail

jobResult = adlaJobClient.job.get(adla, jobId)
while(jobResult.state != JobState.ended):
    print('Job is not yet done, waiting for 3 seconds. Current state: ' +
          jobResult.state.value)
    time.sleep(3)
    jobResult = adlaJobClient.job.get(adla, jobId)

print('Job finished with result: ' + jobResult.result.value)

Répertorier les pipelines et les récurrences

Selon que vos travaux aient des métadonnées de pipeline ou de récurrence attachées, vous pouvez lister les pipelines et les récurrences.

pipelines = adlaJobClient.pipeline.list(adla)
for p in pipelines:
    print('Pipeline: ' + p.name + ' ' + p.pipelineId)

recurrences = adlaJobClient.recurrence.list(adla)
for r in recurrences:
    print('Recurrence: ' + r.name + ' ' + r.recurrenceId)

Gérer les stratégies de calcul

L’objet DataLakeAnalyticsAccountManagementClient fournit des méthodes pour gérer les stratégies de calcul d’un compte Data Lake Analytics.

Répertorier les stratégies de calcul

Le code suivant récupère une liste de stratégies de calcul pour un compte Data Lake Analytics.

policies = adlaAcctClient.compute_policies.list_by_account(rg, adla)
for p in policies:
    print('Name: ' + p.name + 'Type: ' + p.object_type + 'Max AUs / job: ' +
          p.max_degree_of_parallelism_per_job + 'Min priority / job: ' + p.min_priority_per_job)

Créer une stratégie de calcul

Le code suivant crée une stratégie de calcul pour un compte Data Lake Analytics, en définissant le nombre maximal d’unités d’utilisation disponibles pour l’utilisateur spécifié sur 50 et la priorité minimale du travail sur 250.

userAadObjectId = "3b097601-4912-4d41-b9d2-78672fc2acde"
newPolicyParams = CreateOrUpdateComputePolicyParameters(
    userAadObjectId, "User", 50, 250)
adlaAcctClient.compute_policies.create_or_update(
    rg, adla, "GaryMcDaniel", newPolicyParams)

Étapes suivantes

  • Last updated on