Appel des API AWS Glue dans Python

Notez que les API de ressource Boto 3 ne sont pas encore disponibles pour AWS Glue. Actuellement, seules les API client Boto 3 peuvent être utilisées.

Noms d'API AWS Glue dans Python

Les noms d'API AWS Glue dans Java et autres langages de programmation sont généralement CamelCased. Cependant, lors d'un appel à partir de Python, ces noms génériques sont modifiés en minuscules, avec les différentes parties du nom séparées par des traits de soulignement afin de les rendre plus proches du langage Python. Dans la AWS Glue APIdocumentation de référence, ces noms « pythoniens » sont affichés entre parenthèses après les noms génériques CamelCased.

Cependant, même si les noms d'API AWS Glue eux-mêmes sont transformés en minuscules, leurs noms de paramètres restent en majuscule. Il est important de s'en souvenir, car les paramètres doivent être transmis par nom lors de l'appel des API AWS Glue, comme décrit dans la section suivante.

Transmission de paramètres Python et accès à ces paramètres dans AWS Glue

Dans les appels Python d'API AWS Glue, il est préférable de transmettre les paramètres explicitement par nom. Par exemple :

job = glue.create_job(Name='sample', Role='Glue_DefaultRole',
                      Command={'Name': 'glueetl',
                               'ScriptLocation': 's3://my_script_bucket/scripts/my_etl_script.py'})

Il est utile de comprendre que Python crée un dictionnaire de tuplets nom/valeur que vous spécifiez comme arguments de script ETL d'un Structure Job ou JobRun structure. Boto 3 les transmet à AWS Glue au format JSON par le biais d'un appel d'API REST. Cela signifie que vous ne pouvez pas vous fier à l'ordre des arguments lorsque vous y accédez dans votre script.

Par exemple, supposons que vous lanciez un JobRun dans un gestionnaire Lambda Python, et que vous souhaitiez spécifier plusieurs paramètres. Votre code peut alors se présenter comme suit :

from datetime import datetime, timedelta

client = boto3.client('glue')

def lambda_handler(event, context):
  last_hour_date_time = datetime.now() - timedelta(hours = 1)
  day_partition_value = last_hour_date_time.strftime("%Y-%m-%d")
  hour_partition_value = last_hour_date_time.strftime("%-H")

  response = client.start_job_run(
               JobName = 'my_test_Job',
               Arguments = {
                 '--day_partition_key':   'partition_0',
                 '--hour_partition_key':  'partition_1',
                 '--day_partition_value':  day_partition_value,
                 '--hour_partition_value': hour_partition_value } )

Pour accéder à ces paramètres de manière fiable dans votre script ETL, spécifiez-les par nom à l'aide de la fonction AWS Glue getResolvedOptions et accédez-y depuis le dictionnaire obtenu :

import sys
from awsglue.utils import getResolvedOptions

args = getResolvedOptions(sys.argv,
                          ['JOB_NAME',
                           'day_partition_key',
                           'hour_partition_key',
                           'day_partition_value',
                           'hour_partition_value'])
print "The day partition key is: ", args['day_partition_key']
print "and the day partition value is: ", args['day_partition_value']

Si vous souhaitez transmettre un argument qui est une chaîne JSON imbriquée, pour préserver la valeur du paramètre lorsqu'elle est transmise à votre tâche ETL AWS Glue, vous devez coder la chaîne de paramètre avant de démarrer l'exécution de la tâche, puis décoder la chaîne de paramètre avant de la référencer votre script de tâche. Prenons l'exemple de la la chaîne d'arguments suivante :

glue_client.start_job_run(JobName = "gluejobname", Arguments={
"--my_curly_braces_string": '{"a": {"b": {"c": [{"d": {"e": 42}}]}}}'
})

Pour transférer ce paramètre correctement, vous devez coder l'argument sous la forme d'une chaîne codée en Base64.

import base64
...
sample_string='{"a": {"b": {"c": [{"d": {"e": 42}}]}}}'
sample_string_bytes = sample_string.encode("ascii")

base64_bytes = base64.b64encode(sample_string_bytes) 
base64_string = base64_bytes.decode("ascii") 
...
glue_client.start_job_run(JobName = "gluejobname", Arguments={
"--my_curly_braces_string": base64_bytes})
...
sample_string_bytes = base64.b64decode(base64_bytes) 
sample_string = sample_string_bytes.decode("ascii") 
print(f"Decoded string: {sample_string}") 
...

Exemple : Créer et exécuter une tâche

L'exemple suivant montre comment appeler les API AWS Glue avec Python, pour créer et exécuter une tâche ETL.

Pour créer et exécuter une tâche

1. Créez une instance du client AWS Glue :

   import boto3
   glue = boto3.client(service_name='glue', region_name='us-east-1',
                 endpoint_url='https://glue.us-east-1.amazonaws.com')

2. Crée une tâche. Vous devez utiliser glueetl comme nom de la commande ETL, comme indiqué dans le code suivant :

   myJob = glue.create_job(Name='sample', Role='Glue_DefaultRole',
                             Command={'Name': 'glueetl',
                                      'ScriptLocation': 's3://my_script_bucket/scripts/my_etl_script.py'})

3. Démarrez une nouvelle exécution de la tâche que vous avez créée dans l'étape précédente :

   myNewJobRun = glue.start_job_run(JobName=myJob['Name'])

4. Obtenez l'état de la tâche :

   status = glue.get_job_run(JobName=myJob['Name'], RunId=myNewJobRun['JobRunId'])

5. Affichez l'état actuel de l'exécution de la tâche :

   print(status['JobRun']['JobRunState'])