Découvrir le cloud avec AWS en développant un data pipeline — Partie 2

Dans la première partie de ce tutoriel, nous avons présenté brièvement les principes d’un service de cloud computing comme AWS et certains de ses avantages puis avons fait nos premiers pas dans la console. Nous avons ensuite découvert l’un des services les plus emblématiques d’AWS, Amazon S3, service de stockage orienté objet, à travers la création de deux buckets que nous utiliserons dans cet exercice pour stocker nos données de staging mais aussi et surtout les données de notre datalake.

Dans cette deuxième partie, nous développerons nos premières lignes de code sur AWS Lambda, tirerons parti des notifications d’Amazon S3 et ferons nos premiers pas dans IAM, service essentiel d’AWS. Tout cela, depuis la console AWS.

AWS Lambda

Le service AWS Lambda vous permet d’exécuter du code sans avoir à déployer ou gérer des serveurs.
Il vous suffit de développer le code de votre fonction selon le langage de votre choix parmi Node.js, Python, .NET, Go, Java ou Ruby (liste susceptible d’évoluer) et le service s’occupe de l’environnement d’exécution et de la mise à l’échelle.
Votre lambda s’exécutera selon un ou des déclencheurs, appelés events, et passés en argument de votre fonction Lambda. Ces déclencheurs peuvent être manuels, planifiés, lancés depuis vos applicatifs ou encore basés sur les services AWS (arrivée d’un fichier dans un bucket S3, mise à jour d’une table dans DynamoDB, une requête HTTP depuis API Gateway,…).

L’un des intérêts ici est que vous n’êtes facturé que pour le temps effectif de consommation du service, lorsque votre code tourne.

Créons notre première fonction Lambda.
Celle-ci sera déclenchée lors du dépôt d’un fichier de données sur notre bucket S3 de staging.
Notre fonction décompressera notre fichier et le déposera dans le bucket correspondant au datalake. Ce seront nos données “brutes”.

1. Rendez-vous sur le service Lambda.

2. Cliquez sur Create function.

3. Trois possibilités : Créer votre fonction à partir de zéro, depuis un modèle de code et de préréglages de configuration ou selon un exemple d’application du référentiel AWS Serverless Application Repository. Nous allons créer notre modèle à partir de zéro, conservez donc la valeur par défaut.

4. Donnez un nom à votre fonction, par exemple decompress. Puis, choisissez l’environnement d’exécution de la fonction, nous écrirons notre code en Python, choisissez donc Python 3.8.

5. Nous avons ensuite la possibilité de gérer les autorisations de notre fonction Lambda. Trois possibilités :

• Créer un nouveau rôle avec les autorisations Lambda de base en laissant AWS créer ce rôle pour nous ;
• Utiliser un rôle existant, créé par nos soins ;
• Créer un nouveau rôle à partir de la stratégie AWS templates.

Nous souhaitons, outre les autorisations Lambda de base, comme l’accès au logging Amazon CloudWatch pour monitorer notre fonction, donner accès à nos buckets S3 pour y lire nos données, les décompresser et les copier dans notre datalake.
Nous allons donc créer notre propre rôle et c’est sur le service IAM d’AWS que cela se passe. Petit détour.

AWS IAM

IAM est l’un des services les plus importants d’AWS, il est central.
IAM vous permet de gérer l’accès aux ressources AWS. Vous pouvez notamment y créer et gérer les accès des utilisateurs de votre projet, en y associant notamment des rôles, eux même constitués d’autorisations, aussi appelées policies (Understanding how IAM works) pour en savoir plus).

Les ressources que vous créerez (fonction Lambda, job Glue, …) seront également associés à des rôles IAM, on dit qu’ils assument ces rôles. Cela vous permet de contrôler de manière très fine les autorisations de ces ressources.
Vous pouvez, par exemple, attribuer à votre fonction Lambda les droits d’accéder aux données de certains buckets, uniquement en lecture, ou encore les droits de déclencher l’exécution d’un job Glue, les informations du statut de celui-ci, etc.
IAM est un service proposé gratuitement. Vous ne serez facturé que pour l’utilisation des autres services AWS.

Reprenons notre exercice et créons un rôle pour notre fonction Lambda

1. Rendez-vous sur le service IAM.

2. Cliquez sur Roles puis Create Role.

3. Gardez Service AWS comme type d’entité de confiance. Puis, Lambda comme cas d’utilisation. Cela permet de spécifier que le rôle ne peut être attribué qu’à des fonctions lambdas.

Cliquez enfin sur Suivant.

4. Vous devez maintenant définir les autorisations de ce rôle parmi une longue liste d’autorisations prédéfinies et/ou créées par vos soins.
Ajoutons tout d’abord la policy AWSLambdaBasicExecutionRole pour bénéficier des autorisations Lambda de base évoquées précédemment.

5. Nous souhaitons également accéder aux données des buckets S3. Il existe notamment un rôle, AmazonS3FullAccess, qui nous permettrait d’y accéder. Je vous propose cependant de créer notre propre Policy, une série d’autorisations, ou de refus dans certains cas, d’accès à telle ou telle ressource, afin de bénéficier ici uniquement des accès nécessaires sur les buckets qui nous concernent.
Cliquez sur Create policy.

Vous arrivez sur l’éditeur de création de policy. Deux options : l’éditeur graphique ou l’édition de policy en JSON.

6. L’éditeur graphique semble plus accessible, cependant, il est intéressant de voir à quoi peut ressembler une Policy au format JSON. D’autant que c’est souvent dans ce format que vous aurez affaire à ces dernières. Il est donc important de savoir en analyser la construction.
Cliquez donc sur l’onglet JSON.
Copiez-y la policy suivante, en pensant à remplacer par le nom de vos buckets où cela est mentionné, et décrivons la :

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "ListObjectsInBucket",
            "Effect": "Allow",
            "Action": [
                "s3:ListBucket"
            ],
            "Resource": [
                "arn:aws:s3:::",
                "arn:aws:s3:::"
            ]
        },
        {
            "Sid": "AllObjectActions",
            "Effect": "Allow",
            "Action": "s3:*Object",
            "Resource": [
                "arn:aws:s3:::/*",
                "arn:aws:s3:::/*"
            ]
        }
    ]
}

1 : Version spécifie les règles de syntaxe de langage qui doivent être utilisées pour traiter une policy. La version 2012–10–17 est la dernière version et gère notamment les variables.
2 : Nous déclarons une liste des règles/statements
3 : Le Sid vous permet de nommer vos règles pour les identifier.
4 : Effect: Allow pour autoriser, Deny pour restreindre l’accès à une resource.
5 : Ici nous autorisons l’action s3:ListBucket (permet de lister les objets d’un bucket) sur les buckets…
6 : staging-ddp et datalake-ddp
7 : s3:*Object permet d’effectuer toutes les actions possibles du service s3 se terminant par Object (GetObject, ListObject) s’appliquant à :
8 : toutes les ressources ou objet des buckets staging-ddp et datalake-ddp, encore une fois grâce au symbole *
Un conseil : de manière générale, appliquez le principe du moindre privilège.

7. Cliquez sur Review policy. Vous avez là un récapitulatif de votre policy. Nommez là et décrivez là.

8. Cliquez sur Create policy. Vous devriez avoir un message vous informant du succès de la création de celle-ci.

9. Retournez à la création du rôle.

10. Rafraichissez la liste et sélectionnez votre nouvelle création.

11. Cliquez sur Suivant. Vous pouvez définir des tags. Un ensemble de clés-valeurs que vous définissez qui vous permettent de retrouver rapidement vos ressources. Par exemple, le nom de votre projet, le référent…

12. Suivant.

13. Vous avez là également un récapitulatif de votre rôle. Vérifiez, nommez, décrivez et cliquez sur Create role.

La rédaction de policies peut sembler difficile et assez longue. Heureusement, vous disposez d’un générateur en ligne pour vous aider.

Parfait, nous pouvons retourner à la création de notre fonction Lambda.

1. Le rôle que vous venez de créer devrait apparaître dans la liste déroulante, sinon, rafraîchissez les données. Sélectionnez le.

2. Cliquez sur Create function.

3. Petit tour du propriétaire.

Vous avez tout d’abord trois onglets :

• Configuration, avec notamment :

— le Designer, qui vous permet de définir les déclencheurs de votre fonction Lambda mais aussi les Layers, du code supplémentaire, sous la forme de couches (layers). Une couche est une archive ZIP qui contient des bibliothèques, un environnement d’exécution personnalisé ou d’autres dépendances que vous pourrez utiliser dans le code de votre fonction ;

— un éditeur pour le code de votre fonction ;

— la possibilité d’ajouter des tags ;

— des variables d’environnement ;

— et d’autres éléments de configuration tels que les ressources allouées à votre environnement d’exécution, le timeout de la fonction, des paramètres réseau…

• Permission pour la gestion des permissions;
• Monitoring, où vous retrouverez divers metrics sur les différentes exécutions de votre fonction. Le monitoring s’appuie sur le service CloudWatch d’AWS.

4. Nous souhaitons que notre fonction décompresse les fichiers de données et les déplace dans notre datalake, et ce dès leur arrivée dans notre bucket de staging. Commençons par le commencement, le déclenchement de notre fonction.
Dans le Designer, cliquez sur Ajoutez un déclencheur (Add trigger).

a. Dans la liste déroulante, recherchez et sélectionnez le service S3.

b. Puis, sélectionnez votre Bucket de staging.

c. Dans Event type, gardez All object create events. Nous pourrions, par exemple, déclencher notre fonction lors de la suppression d’un objet dans notre bucket.

d. Dans Prefix, saisissez in/. De telle manière, nous déclencherons notre fonction pour les fichiers créés/ajoutés dans le ‘dossier’ in/ de notre bucket (Difference between prefixes and nested folders). Nous aurions pu ne rien mettre mais cela nous laisse la possibilité d’utiliser notre bucket de staging pour autre chose, comme nous le verrons très vite.

e. Dans Suffix, saisissez .zip, ainsi nous ne déclencherons notre fonction que si le fichier est un fichier ayant l’extension .zip.
AWS nous informe qu’il ajoutera les autorisations nécessaires pour que le service S3 appelle notre fonction Lambda depuis ce déclencheur.

f. Vous pouvez activer ou désactiver les déclencheurs, nous allons le laisser activé.
Cliquez sur Add.

5. Passons à l’édition du code de notre fonction.

Quelque soit le langage choisi, le code de votre fonction Lambda est construit avec une fonction appelée handler que le service Lambda appelle en lui passant en paramètre deux arguments :

• context, contient des propriétés sur la fonction telles que le nom et la version de la fonction, la quantité de mémoire allouée à la fonction, etc.
• event, pour transmettre les paramètres d’appel de la fonction lambda, souvent sous la forme d’un document json (ou un dict Python). Son contenu diffère selon le déclencheur.

Pour un évènement S3, il contient par exemple le timestamp de l’évènement, le bucket, la key de l’objet ou encore l’eTag, un hachage de l’objet qui permet de vérifier l’intégrité de l’objet.

Le cas échéant, la fonction peut renvoyer une valeur. Ce qu’il advient de la valeur renvoyée dépend du type d’appel utilisé.

6. Ajoutez le code suivant dans l’éditeur, là aussi en pensant à remplacer par le nom de votre bucket où il le faut.

import boto3
import os
from zipfile import ZipFile

def lambda_handler(event, context):
    
    s3 = boto3.resource('s3')

    try:
        src_bucket = event['Records'][0]['s3']['bucket']['name']
        src_key = event['Records'][0]['s3']['object']['key']
        filename = os.path.basename(src_key)
        table_name = os.path.splitext(filename)[0]
        target_bucket = ''
        files_list = []

        s3.meta.client.download_file(src_bucket, src_key, f'/tmp/{filename}')
        
        with ZipFile(f'/tmp/{filename}', 'r') as zipObj:
            zipObj.extractall('/tmp/unzipped')
        
        
        for file in os.listdir('/tmp/unzipped'):
            s3.meta.client.upload_file(f'/tmp/unzipped/{file}', target_bucket, f'raw/{table_name}/{file}')
            files_list.append(file)
    except Exception as e:
        s3.meta.client.copy({'Bucket': src_bucket, 'Key': src_key}, src_bucket, f'out/{filename}')
    finally:
        s3.Object(src_bucket, src_key).delete()
    
    return files_listL’éditeur permet de tester vos fonctions en définissant un objet passé en event à votre handler, nous ne l’utiliserons pas ici mais il peut-être utile.

7. Vous pouvez ajouter un tag, comme nous l’avons fait lors de la création de notre rôle, c’est une bonne pratique. Puis enregistrez votre fonction en cliquant sur Save, en haut à droite de la page.
Bien, testons maintenant notre fonction.

8. Téléchargez le fichier de données en cliquant ici, ils s’agit d’une liste de candidats arbitraire.

9. Retournez sur la page du service S3 et cliquez sur le bucket de staging.

10. Cliquez sur Create folder, nommez le in. Cliquez sur Save.

11. Rendez-vous dans le dossier fraîchement créé et cliquez sur Upload. Cliquez sur Add files pour ajouter le fichier candidats téléchargé plus haut. Puis sur Upload.

12. Ceci aura pour effet de déclencher notre fonction Lambda. Rendez-vous dans le bucket datalake. Après quelques secondes, vous devriez trouver un dossier du nom du fichier téléchargé (correspondant au nom de notre table) sous le préfixe raw/, contenant la liste des fichiers décompressés.

Dans la partie suivante, nous travaillerons sur le cœur de notre data pipeline, le job de transformation des données. Il s’appuiera sur Glue, service qui nous permettra également d’analyser ces données afin d’en déduire des tables structurées prêtes à l’exploitation.