Llamada a las API de AWS Glue en Python - AWS Glue

Llamada a las API de AWS Glue en Python

Tenga en cuenta que las API de recurso de Boto 3 aún no están disponibles para AWS Glue. Actualmente, solo se pueden usar las API de cliente de Boto 3.

AWS GlueNombres de API de en Python

Los nombres de API de AWS Glue en Java y otros lenguajes de programación suelen ser CamelCased. Sin embargo, al recibir la llamada de Python, estos nombres genéricos se cambian a minúsculas, con las partes del nombre separadas por caracteres de guion bajo para que sean "más Python". En la documentación de referencia API de AWS Glue, estos nombres Python aparecen enumerados entre paréntesis después de los nombres CamelCased genéricos.

Sin embargo, aunque los propios nombres de API de AWS Glue cambian a minúsculas, los nombres de sus parámetros permanecen en mayúsculas. Es importante recordar esto, ya que los parámetros deben transferirse por nombre al llamar a las API de AWS Glue, tal como se describe en la sección siguiente.

Transferencia y acceso a los parámetros de Python en AWS Glue

En las llamadas de Python a las API de AWS Glue, es mejor transferir los parámetros explícitamente por nombre. Por ejemplo:

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

Resulta útil para comprender que Python crea un diccionario de las tuplas nombre/valor que especifica como argumentos para un script de ETL en Estructura Job o Estructura JobRun. Boto 3 los transfiere a continuación a AWS Glue en formato JSON a través de una llamada a la API REST. Esto significa que no puede depender del orden de los argumentos al obtener acceso a ellos en su script.

Por ejemplo, supongamos que inicia JobRun en una función del controlador de Lambda de Python y desea especificar varios parámetros. Su código podría tener un aspecto similar al siguiente:

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 } )

Para obtener acceso a estos parámetros de forma fiable en su script ETL, especifíquelos por nombre mediante la función AWS Glue de getResolvedOptions y, a continuación, obtenga acceso a ellos desde el diccionario resultante:

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 desea transferir un argumento que es una cadena JSON anidada, para preservar el valor del parámetro a medida que se transfiere a su trabajo de ETL de AWS Glue, debe codificar la cadena de parámetros antes de iniciar la ejecución del trabajo y, a continuación, decodificar la cadena de parámetros antes de hacer referencia a su script de trabajo. Por ejemplo, tenga en cuenta las siguientes cadenas de argumentos:

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

Para transferir este parámetro correctamente, debe codificar el argumento como una cadena codificada 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}") ...

Ejemplo: crear y ejecutar un flujo de trabajo

En el siguiente ejemplo se muestra cómo llamar a las API de AWS Glue con Python, para crear y ejecutar un flujo de trabajo de ETL.

Para crear y ejecutar un flujo de trabajo

  1. Cree una instancia del cliente de 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. Crear un flujo de trabajo. Debe utilizar glueetl como nombre del comando de ETL, tal como se muestra en el siguiente código:

    myJob = glue.create_job(Name='sample', Role='Glue_DefaultRole', Command={'Name': 'glueetl', 'ScriptLocation': 's3://my_script_bucket/scripts/my_etl_script.py'})
  3. Inicie una nueva ejecución del flujo de trabajo que creó en el paso anterior:

    myNewJobRun = glue.start_job_run(JobName=myJob['Name'])
  4. Obtenga el estado del flujo de trabajo:

    status = glue.get_job_run(JobName=myJob['Name'], RunId=myNewJobRun['JobRunId'])
  5. Imprima el estado actual de la ejecución del flujo de trabajo:

    print status['JobRun']['JobRunState']