SDK Azure Data Lake Storage Gen2 pour Python

Système de fichiers hiérarchique pour les charges de travail d'analyse de données massives.

Installation

pip install azure-storage-file-datalake azure-identity

Variables d'environnement

AZURE_STORAGE_ACCOUNT_URL=https://<account>.dfs.core.windows.net  # Obligatoire pour toutes les méthodes d'authentification
AZURE_TOKEN_CREDENTIALS=prod # Obligatoire uniquement si DefaultAzureCredential est utilisé en production
</account>

Authentification et cycle de vie

🔑 Deux règles s'appliquent à chaque exemple de code ci-dessous :

1. Privilégiez DefaultAzureCredential. Il fonctionne localement (Azure CLI / VS Code / Developer CLI) et dans Azure (identité managée, identité de charge de travail) sans modification de code. Évitez les chaînes de connexion, les comptes/clés API — ils contournent l'audit Entra et la rotation.
   • Développement local : DefaultAzureCredential fonctionne tel quel.
   • Production : définissez AZURE_TOKEN_CREDENTIALS=prod (ou AZURE_TOKEN_CREDENTIALS=<specific_credential></specific_credential>) pour limiter la chaîne d'identification aux identifiants sécurisés pour la production.
2. Enveloppez chaque client dans un gestionnaire de contexte afin que les transports HTTP, les sockets et les caches de jetons soient libérés de manière déterministe :
   • Synchrone : with <client>(...) as client:</client>
   • Asynchrone : async with <client>(...) as client:</client> et async with DefaultAzureCredential() as credential: (de azure.identity.aio)

Les extraits de code peuvent abréger cette configuration, mais le code de production doit toujours respecter les deux règles.

from azure.identity import DefaultAzureCredential, ManagedIdentityCredential
from azure.storage.filedatalake import DataLakeServiceClient

# Développement local : DefaultAzureCredential. Production : définissez AZURE_TOKEN_CREDENTIALS=prod ou AZURE_TOKEN_CREDENTIALS=<specific_credential>
credential = DefaultAzureCredential(require_envvar=True)
# Ou utilisez un identifiant spécifique directement en production :
# Voir https://learn.microsoft.com/python/api/overview/azure/identity-readme?view=azure-python#credential-classes
# credential = ManagedIdentityCredential()
account_url = "https://<account>.dfs.core.windows.net"

with DataLakeServiceClient(account_url=account_url, credential=credential) as service_client:
    # Utilisez service_client ici (voir les sections suivantes pour les opérations)
    ...
</account></specific_credential>

Hiérarchie des clients

Opérations sur le système de fichiers

# Créer un système de fichiers (conteneur)
file_system_client = service_client.create_file_system("myfilesystem")

# Obtenir l'existant
file_system_client = service_client.get_file_system_client("myfilesystem")

# Supprimer
service_client.delete_file_system("myfilesystem")

# Lister les systèmes de fichiers
for fs in service_client.list_file_systems():
    print(fs.name)

Opérations sur les répertoires

file_system_client = service_client.get_file_system_client("myfilesystem")

# Créer un répertoire
directory_client = file_system_client.create_directory("mydir")

# Créer des répertoires imbriqués
directory_client = file_system_client.create_directory("path/to/nested/dir")

# Obtenir le client de répertoire
directory_client = file_system_client.get_directory_client("mydir")

# Supprimer le répertoire
directory_client.delete_directory()

# Renommer/déplacer le répertoire
directory_client.rename_directory(new_name="myfilesystem/newname")

Opérations sur les fichiers

Téléchargement de fichier (Upload)

# Obtenir le client de fichier
file_client = file_system_client.get_file_client("path/to/file.txt")

# Télécharger depuis un fichier local
with open("local-file.txt", "rb") as data:
    file_client.upload_data(data, overwrite=True)

# Télécharger des octets
file_client.upload_data(b"Hello, Data Lake!", overwrite=True)

# Ajouter des données (pour les gros fichiers)
file_client.append_data(data=b"chunk1", offset=0, length=6)
file_client.append_data(data=b"chunk2", offset=6, length=6)
file_client.flush_data(12)  # Valider les données

Téléchargement de fichier (Download)

file_client = file_system_client.get_file_client("path/to/file.txt")

# Télécharger tout le contenu
download = file_client.download_file()
content = download.readall()

# Télécharger vers un fichier
with open("downloaded.txt", "wb") as f:
    download = file_client.download_file()
    download.readinto(f)

# Télécharger une plage
download = file_client.download_file(offset=0, length=100)

Supprimer un fichier

file_client.delete_file()

Lister le contenu

# Lister les chemins (fichiers et répertoires)
for path in file_system_client.get_paths():
    print(f"{'DIR' if path.is_directory else 'FILE'}: {path.name}")

# Lister les chemins dans un répertoire
for path in file_system_client.get_paths(path="mydir"):
    print(path.name)

# Liste récursive
for path in file_system_client.get_paths(path="mydir", recursive=True):
    print(path.name)

Propriétés des fichiers/répertoires

# Obtenir les propriétés
properties = file_client.get_file_properties()
print(f"Taille : {properties.size}")
print(f"Dernière modification : {properties.last_modified}")

# Définir les métadonnées
file_client.set_metadata(metadata={"processed": "true"})

Contrôle d'accès (ACL)

# Obtenir l'ACL
acl = directory_client.get_access_control()
print(f"Propriétaire : {acl['owner']}")
print(f"Autorisations : {acl['permissions']}")

# Définir l'ACL
directory_client.set_access_control(
    owner="user-id",
    permissions="rwxr-x---"
)

# Mettre à jour les entrées ACL
from azure.storage.filedatalake import AccessControlChangeResult
directory_client.update_access_control_recursive(
    acl="user:user-id:rwx"
)

Client asynchrone

from azure.storage.filedatalake.aio import DataLakeServiceClient
from azure.identity.aio import DefaultAzureCredential

async def datalake_operations():
    async with DefaultAzureCredential() as credential:
        async with DataLakeServiceClient(
            account_url="https://<account>.dfs.core.windows.net",
            credential=credential
        ) as service_client:
            file_system_client = service_client.get_file_system_client("myfilesystem")
            file_client = file_system_client.get_file_client("test.txt")

            await file_client.upload_data(b"async content", overwrite=True)

            download = await file_client.download_file()
            content = await download.readall()

import asyncio
asyncio.run(datalake_operations())
</account>

Meilleures pratiques

1. Choisissez synchrone OU asynchrone et restez cohérent. Ne mélangez pas les clients synchrones azure.storage.filedatalake avec les clients asynchrones azure.storage.filedatalake.aio dans le même chemin d'appel. Choisissez un mode par module.
2. Utilisez toujours des gestionnaires de contexte pour les clients et les identifiants asynchrones. Enveloppez chaque client dans with DataLakeServiceClient(...) as client: (synchrone) ou async with DataLakeServiceClient(...) as client: (asynchrone). Pour DefaultAzureCredential asynchrone de azure.identity.aio, utilisez également async with credential: afin que les jetons et les transports soient nettoyés correctement.
3. Utilisez DefaultAzureCredential pour une authentification portable entre le développement local et Azure (évitez les chaînes de connexion / clés API lorsque cela est possible).
4. Utilisez un espace de noms hiérarchique pour la sémantique du système de fichiers
5. Utilisez append_data + flush_data pour les téléchargements de gros fichiers
6. Définissez les ACL au niveau du répertoire et héritez-les aux enfants
7. Utilisez le client asynchrone pour les scénarios à haut débit
8. Utilisez get_paths avec recursive=True pour une liste complète des répertoires
9. Définissez les métadonnées pour les attributs de fichiers personnalisés
10. Envisagez l'API Blob pour les cas d'utilisation simples de stockage d'objets