Étape 2. Implémenter la logique de transformation

Note

Les transformations visuelles personnalisées ne prennent en charge que les scripts Python. Scala n'est pas pris en charge.

Pour ajouter le code qui implémente la fonction définie par le fichier de configuration .json, il est recommandé de placer le fichier Python au même endroit que le fichier .json, avec le même nom mais avec l'extension « .py ». AWS Glue Studio associe automatiquement les fichiers .json et .py afin que vous n'ayez pas besoin de spécifier le chemin du fichier Python dans le fichier de configuration.

Dans le fichier Python, ajoutez la fonction déclarée, avec les paramètres nommés configurés, et enregistrez-la pour qu'elle soit utilisée dans DynamicFrame. Voici un exemple de fichier Python :

from awsglue import DynamicFrame

# self refers to the DynamicFrame to transform, 
# the parameter names must match the ones defined in the config
# if it's optional, need to provide a default value
def myTransform(self, email, phone, age=None, gender="", 
                      country="", promotion=False):
   resulting_dynf = # do some transformation on self 
   return resulting_dynf
   
DynamicFrame.myTransform = myTransform
        

Il est recommandé d'utiliser un AWS Glue bloc-notes pour le moyen le plus rapide de développer et de tester le code python. Voir Commencer à utiliser les blocs-notes dans AWS Glue Studio.

Pour illustrer l'implémentation de la logique de transformation, la transformation visuelle personnalisée présentée dans l'exemple ci-dessous permet de filtrer les données entrantes afin de ne conserver que les données relatives à un État américain en particulier. Le fichier .json contient le paramètre custom_filter_state pour functionName ainsi que deux arguments (« state » et « colName » avec le type « str »).

L'exemple de fichier de configuration .json est le suivant :

{
"name": "custom_filter_state",
"displayName": "Filter State",
"description": "A simple example to filter the data to keep only the state indicated.",
"functionName": "custom_filter_state",
"parameters": [
   {
    "name": "colName",
    "displayName": "Column name",
    "type": "str",
    "description": "Name of the column in the data that holds the state postal code"
   },
   {
    "name": "state",
    "displayName": "State postal code",
    "type": "str",
    "description": "The postal code of the state whole rows to keep"
   }   
  ]
}
        

Pour implémenter le script compagnon en Python

1. Démarrez un AWS Glue bloc-notes et exécutez la cellule initiale fournie pour démarrer la session. L'exécution de la cellule initiale crée les composants de base nécessaires.

2. Créez une fonction qui procède au filtrage, comme décrit dans l'exemple, et enregistrez-la sur DynamicFrame. Copiez le code ci-dessous et collez-le dans une cellule du AWS Glue carnet.

   from awsglue import DynamicFrame
   
   def custom_filter_state(self, colName, state):
       return self.filter(lambda row: row[colName] == state)
   
   DynamicFrame.custom_filter_state = custom_filter_state                    
                   

3. Créez ou chargez des exemples de données pour tester le code dans la même cellule ou dans une nouvelle cellule. Si vous ajoutez les exemples de données dans une nouvelle cellule, n'oubliez pas d'exécuter la cellule. Par exemple :

   # A few of rows of sample data to test
   data_sample = [
       {"state": "CA", "count": 4},
       {"state": "NY", "count": 2},
       {"state": "WA", "count": 3}    
   ]
   df1 = glueContext.sparkSession.sparkContext.parallelize(data_sample).toDF()
   dynf1 = DynamicFrame.fromDF(df1, glueContext, None)                    
                   

4. Test pour valider « custom_filter_state » avec différents arguments :

   

5. Après avoir effectué plusieurs tests, enregistrez le code avec l'extension .py et attribuez au fichier .py le même nom que le fichier .json. Les fichiers .py et .json doivent se trouver dans le même dossier de transformations.

   Copiez le code suivant, collez-le dans un fichier et renommez celui-ci avec l'extension de fichier .py.

   from awsglue import DynamicFrame
   
   def custom_filter_state(self, colName, state):
       return self.filter(lambda row: row[colName] == state)
   
   DynamicFrame.custom_filter_state = custom_filter_state                    
                   

6. Entrée AWS Glue Studio, ouvrez une tâche visuelle et ajoutez-y la transformation en la sélectionnant dans la liste des transformations disponibles.

   Pour réutiliser cette transformation dans le code d'un script Python, ajoutez le chemin d'accès Amazon S3 au fichier .py dans la tâche située sous « Referenced files path » (Chemin des fichiers référencés) et, dans le script, importez le nom du fichier Python (sans l'extension) en l'ajoutant en haut du fichier. Par exemple : import <nom du fichier (sans l'extension)>